"""Root-scoped views over the central PDF catalog."""
from __future__ import annotations
import copy
import json
from collections import Counter
from collections.abc import Callable, Collection, Iterator, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, NamedTuple
from .arxiv import arxiv_id
from .catalog import (
_DOCUMENT_SNAPSHOT_UNSET,
CATALOG_SCHEMA_VERSION,
LEGACY_REGISTRY_FILENAME,
LOCKED_METADATA_FIELDS_KEY,
PdfCatalog,
file_facts_changed,
)
from .classification import IDENTIFICATION_GROUPS, identification_group
from .files import sha256_file
from .identify_settings import IdentifySettings
from .identity import IDENTITY_KEY
from .readiness import METADATA_STALE_KEY, stored_bibcode_matches_file
from .search import document_matches, fold_document, record_search_document
from .upgrade_utils import is_arxiv_doi
if TYPE_CHECKING:
from .rename_settings import RenameSettings
REGISTRY_FILENAME = LEGACY_REGISTRY_FILENAME
[docs]
class RecordFact(NamedTuple):
"""Hold the record values that view and status counts need.
The fields are the canonical library root, the catalog location
identifier, the value of ``identification_group(record)``, the saved
metadata source, and the current content digest. A missing source or
digest is an empty string.
"""
root: str
location_id: int
group: str
source: str
sha256: str
document_id: int | None = None
edition: str = ""
DUPLICATE_SAME_FILE = "same_file"
DUPLICATE_SAME_PAPER = "same_paper"
DUPLICATE_LABELS = {
DUPLICATE_SAME_FILE: "Same file",
DUPLICATE_SAME_PAPER: "Same paper",
}
[docs]
def pdf_edition(record: Mapping[str, Any]) -> str:
"""Return ``"preprint"``, ``"published"``, or ``""`` from the PDF's own IDs.
The identity proof lists the identifiers that the PDF states. A journal
DOI marks the published edition. An arXiv ID or arXiv DOI alone marks the
preprint. ADS can link both editions to one paper record.
"""
metadata = record.get("metadata")
proof = metadata.get(IDENTITY_KEY) if isinstance(metadata, Mapping) else None
if not isinstance(proof, Mapping):
return ""
dois = [str(value) for value in proof.get("own_doi") or ()]
if any(not is_arxiv_doi(value) for value in dois):
return "published"
if proof.get("own_arxiv") or dois:
return "preprint"
return ""
[docs]
def duplicate_kinds(keys: Sequence[tuple[str, int | None, str]]) -> list[str]:
"""Return the duplicate kind of each PDF, or ``""`` for a single copy.
Parameters
----------
keys
``(sha256, document_id, edition)`` of each PDF, with the edition from
`pdf_edition`. An empty digest or a `None` document does not group.
Returns
-------
list of str
`DUPLICATE_SAME_FILE` when another PDF has the same bytes. Otherwise
`DUPLICATE_SAME_PAPER` when another PDF links to the same paper record
and is the same edition. An unknown edition matches either edition. A
preprint and its published edition are not duplicates.
"""
digests = Counter(digest for digest, _document, _edition in keys if digest)
editions: dict[int, list[str]] = {}
for _digest, document, edition in keys:
if document is not None:
editions.setdefault(document, []).append(edition)
def same_paper(document: int | None, edition: str) -> bool:
if document is None:
return False
others = list(editions[document])
others.remove(edition)
return any(not edition or not other or other == edition for other in others)
return [
DUPLICATE_SAME_FILE
if digest and digests[digest] > 1
else DUPLICATE_SAME_PAPER
if same_paper(document, edition)
else ""
for digest, document, edition in keys
]
[docs]
def mark_duplicates(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Return the duplicate records with ``duplicate_kind`` and ``duplicate_group``.
Copies of one paper share a group, including copies with the same bytes.
When ADS links a preprint and its published edition to one record, each
edition forms its own group. Copies without a paper record group by bytes.
"""
editions = [pdf_edition(record) for record in records]
kinds = duplicate_kinds(
[
(str(record.get("sha256") or ""), record.get("document_id"), edition)
for record, edition in zip(records, editions, strict=True)
]
)
known: dict[int, set[str]] = {}
for record, edition, kind in zip(records, editions, kinds, strict=True):
if kind and edition and record.get("document_id") is not None:
known.setdefault(record["document_id"], set()).add(edition)
marked = []
for record, edition, kind in zip(records, editions, kinds, strict=True):
if not kind:
continue
document = record.get("document_id")
if document is None:
group = f"file:{record.get('sha256')}"
elif len(known.get(document, ())) > 1:
group = f"paper:{document}:{edition}"
else:
group = f"paper:{document}"
record["duplicate_kind"] = kind
record["duplicate_group"] = group
marked.append(record)
return marked
[docs]
def order_duplicate_groups(records: list[dict[str, Any]]) -> None:
"""Keep each duplicate group together and number the groups, in place.
The first copy of each group keeps its place in the current sort. The
other copies of that group follow it. Groups are numbered from 1 in that
order, in ``duplicate_group_number``.
"""
order: dict[str, int] = {}
for record in records:
order.setdefault(str(record.get("duplicate_group") or ""), len(order))
records.sort(key=lambda record: order[str(record.get("duplicate_group") or "")])
for record in records:
record["duplicate_group_number"] = (
order[str(record.get("duplicate_group") or "")] + 1
)
SCHEMA_VERSION = CATALOG_SCHEMA_VERSION
ADS_METADATA_FIELDS = (
"abstract",
"author",
"author_count",
"bibcode",
"identifier",
"alternate_bibcode",
"citation_count",
"doi",
"page",
"pub",
"title",
"volume",
"year",
"bibstem",
)
NUMERIC_SORT_COLUMNS = frozenset(
{"size_bytes", "mtime_ns", "author_count", "citation_count"}
)
# Counts stored in record metadata. A missing value is unknown, not zero.
STORED_COUNT_KEYS = frozenset({"author_count", "citation_count"})
SORTABLE_COLUMNS = {
key: key
for key in (
"title",
"author",
"year",
"pub",
"bibcode",
"arxiv",
"doi",
"status",
"source",
"filename",
"relative_path",
"size_bytes",
"mtime_ns",
"author_count",
"citation_count",
"first_seen_at",
"updated_at",
"identified_at",
"duplicate_kind",
)
}
[docs]
def stored_count(record: Mapping[str, Any], key: str) -> int | None:
"""Return a saved non-negative count from a catalog record, or `None`.
The record's own field is read first, then its `metadata` mapping. Whole
numbers (`int`) and strings of decimal digits are accepted. Booleans,
negative numbers, floats, and other text, including malformed digit
strings, are treated as unknown, so a missing value is never reported as
zero. The count is the value saved at the last ADS lookup, refresh, or
manual edit, not a live ADS value.
Parameters
----------
record : Mapping[str, Any]
Catalog record from ``PdfRegistry.query_files()``.
key : str
Metadata key, such as `"author_count"` or `"citation_count"`.
Returns
-------
int or None
The saved count, or `None` when no valid count is saved.
"""
metadata = record.get("metadata")
sources = (record, metadata if isinstance(metadata, Mapping) else {})
for source in sources:
value = source.get(key)
if isinstance(value, bool):
continue
if isinstance(value, int) and value >= 0:
return value
if isinstance(value, str):
try:
parsed = int(value.strip())
except ValueError:
continue
if parsed >= 0 and value.strip().isdecimal():
return parsed
return None
__all__ = [
"ADS_METADATA_FIELDS",
"DUPLICATE_LABELS",
"DUPLICATE_SAME_FILE",
"DUPLICATE_SAME_PAPER",
"LOCKED_METADATA_FIELDS_KEY",
"REGISTRY_FILENAME",
"duplicate_kinds",
"mark_duplicates",
"order_duplicate_groups",
"pdf_edition",
"SCHEMA_VERSION",
"SORTABLE_COLUMNS",
"STORED_COUNT_KEYS",
"FileReconciliation",
"LibraryRefresh",
"PdfRegistry",
"PdfRegistryState",
"merge_ads_metadata",
"migrate_sidecar",
"open_registry_state",
"stored_count",
]
[docs]
@dataclass(frozen=True)
class FileReconciliation:
"""Summary of filesystem facts reconciled against one catalog root."""
scanned: int
updated: int
missing: int
[docs]
@dataclass(frozen=True)
class LibraryRefresh:
"""Count local file changes while retaining missing-location history."""
scanned: int
added: int
updated: int
moved: int
restored: int
missing: int
[docs]
class CentralPdfRegistry:
"""Root-scoped view over the user-wide :class:`PdfCatalog`."""
def __init__(self, directory: Path) -> None:
self.directory = Path(directory).expanduser().resolve()
self.catalog = PdfCatalog()
self.root_id = self.catalog.ensure_root(self.directory)
self.catalog.migrate_legacy_root(self.directory)
self.path = self.catalog.path
self._connection = self.catalog._connection
self._facts_cache: tuple[tuple[int, ...], tuple[RecordFact, ...]] | None = None
self._search_cache: (
tuple[tuple[Any, ...], frozenset[tuple[str, int]]] | None
) = None
self._haystack_cache: (
tuple[
tuple[int, ...],
tuple[tuple[int, dict[str, str], dict[str, str]], ...],
]
| None
) = None
[docs]
@property
def schema_version(self) -> int:
"""Return the central catalog schema version."""
return self.catalog.schema_version
[docs]
def count_files(self) -> int:
"""Return the number of current locations in this root."""
return self.catalog.count_files(root=self.directory)
[docs]
def record_facts(self) -> tuple[RecordFact, ...]:
"""Return count facts for every current location in this root.
The result is cached until the catalog change token changes, so
repeated view and status counts parse the metadata JSON once.
"""
token = self.catalog.change_token()
cached = self._facts_cache
if cached is not None and cached[0] == token:
return cached[1]
root = str(self.directory)
facts = tuple(
RecordFact(
root,
int(record["location_id"]),
identification_group(record),
str(record.get("source") or ""),
str(record.get("sha256") or ""),
record.get("document_id"),
pdf_edition(record),
)
for record in self.catalog.query_files(root=self.directory)
)
# Keep the token from before the query. A concurrent write then
# invalidates this entry at the next call.
self._facts_cache = (token, facts)
return facts
[docs]
def search_location_keys(
self,
search: str,
*,
exact: bool = False,
wildcards: bool = False,
fields: Collection[str] = (),
) -> frozenset[tuple[str, int]]:
"""Return ``(root, location_id)`` keys that match the query `search`.
The keys equal the result of the catalog search filter with the same
options. `fields` names the fields that plain terms search. See
`adsfolio.pdf.search.document_matches` for the rules. The last result
is cached until the catalog change token changes.
"""
token = self.catalog.change_token()
request = (token, search, exact, wildcards, tuple(fields))
cached = self._search_cache
if cached is not None and cached[0] == request:
return cached[1]
root = str(self.directory)
keys = frozenset(
(root, location_id)
for location_id, exact_document, folded_document in (
self._search_documents(token)
)
if document_matches(
exact_document if exact else folded_document,
search,
exact=exact,
wildcards=wildcards,
default_fields=fields,
)
)
self._search_cache = (request, keys)
return keys
[docs]
def search_documents(
self,
) -> tuple[tuple[int, dict[str, str], dict[str, str]], ...]:
"""Return exact and folded search documents for current locations.
Each entry holds the location identifier and the outputs of
`record_search_document` and `fold_document`. Do not change them.
"""
return self._search_documents(self.catalog.change_token())
def _search_documents(
self, token: tuple[int, ...]
) -> tuple[tuple[int, dict[str, str], dict[str, str]], ...]:
"""Return search documents, cached by change token."""
cached = self._haystack_cache
if cached is not None and cached[0] == token:
return cached[1]
documents = []
for (
location_id,
relative_path,
metadata_json,
sha256,
) in self.catalog.search_texts(self.directory):
document = record_search_document(relative_path, metadata_json, sha256)
documents.append((location_id, document, fold_document(document)))
result = tuple(documents)
self._haystack_cache = (token, result)
return result
[docs]
def count_ads_documents(self) -> int:
"""Return the number of shared cached ADS documents."""
return self.catalog.count_ads_documents()
[docs]
def reconcile_file_facts(
self,
root: Path,
*,
recursive: bool = True,
cancelled: Callable[[], bool] | None = None,
) -> FileReconciliation:
"""Refresh current filesystem facts without deleting locations."""
records = self.catalog.query_files(root=root)
if not recursive:
records = [
record for record in records if "/" not in record["current_path"]
]
updated = 0
missing = 0
for record in records:
if cancelled is not None and cancelled():
raise InterruptedError("Local refresh cancelled")
path = Path(root) / str(record["current_path"])
try:
stat = path.stat()
except OSError:
missing += 1
continue
if not file_facts_changed(record, stat):
continue
try:
sha256 = sha256_file(path)
except OSError:
missing += 1
continue
self.catalog.observe_file(
root,
str(record["current_path"]),
sha256=sha256,
size_bytes=stat.st_size,
mtime_ns=stat.st_mtime_ns,
device=stat.st_dev,
inode=stat.st_ino,
metadata=None,
status=None,
)
updated += 1
return FileReconciliation(
scanned=len(records),
updated=updated,
missing=missing,
)
[docs]
def refresh_local_files(
self,
*,
recursive: bool = True,
cancelled: Callable[[], bool] | None = None,
) -> LibraryRefresh:
"""Scan local PDFs, update changed facts, and retain missing history.
No ADS queries or file renames are performed. Files with changed facts
are hashed before discovery updates those facts. Incomplete directory
scans raise an error instead of marking unseen paths missing.
"""
before = self.catalog.query_files(root=self.directory, include_missing=True)
if not recursive:
before = [record for record in before if "/" not in record["current_path"]]
reconciled = self.reconcile_file_facts(
self.directory, recursive=recursive, cancelled=cancelled
)
scanned = self.catalog.discover_root(
self.directory, recursive=recursive, cancelled=cancelled
)
after = self.catalog.query_files(root=self.directory)
if not recursive:
after = [record for record in after if "/" not in record["current_path"]]
previous = {
record["location_id"] for record in before if not record["missing_at"]
}
current = {record["location_id"] for record in after}
disappeared = Counter(
record["attachment_id"]
for record in before
if record["location_id"] in previous - current
)
appeared = [record for record in after if record["location_id"] not in previous]
historical = {
record["location_id"] for record in before if record["missing_at"]
}
arrivals = Counter(
record["attachment_id"]
for record in appeared
if record["location_id"] not in historical
)
moved = sum(
min(count, arrivals[attachment])
for attachment, count in disappeared.items()
)
restored = sum(record["location_id"] in historical for record in appeared)
return LibraryRefresh(
scanned=scanned,
added=len(appeared) - moved - restored,
updated=reconciled.updated,
moved=moved,
restored=restored,
missing=sum(disappeared.values()) - moved,
)
[docs]
def upsert_file(
self,
*,
root: Path,
relative_path: str,
current_path: str,
sha256: str,
size_bytes: int,
mtime_ns: int,
status: str,
metadata: Mapping[str, Any] | None = None,
) -> int:
"""Register one current root-relative file location."""
del relative_path
path = Path(current_path)
actual_path = Path(root) / path
device: int | None = None
inode: int | None = None
try:
stat = actual_path.stat()
device = stat.st_dev
inode = stat.st_ino
except OSError:
pass
proof = (metadata or {}).get("_adsfolio_identity")
if (
isinstance(proof, Mapping)
and proof.get("status") == "verified"
and actual_path.is_file()
):
replacement = self.catalog.register_verified_replacement(
root, current_path, {**dict(metadata or {}), "sha256": sha256}
)
if replacement is not None:
return replacement.attachment_id
observation = self.catalog.observe_file(
root,
current_path,
sha256=sha256 or None,
size_bytes=max(0, int(size_bytes)),
mtime_ns=max(0, int(mtime_ns)),
device=device,
inode=inode,
metadata=metadata,
status=status,
)
return observation.attachment_id
[docs]
def find_file_by_sha256(self, sha256: str) -> dict[str, Any] | None:
"""Return the first current location matching a digest."""
records = self.catalog.query_files(root=self.directory, sha256=sha256)
return records[0] if records else None
[docs]
def find_files_by_sha256(self, sha256_values: list[str]) -> list[dict[str, Any]]:
"""Return current locations matching any supplied digest."""
values = {str(value).strip() for value in sha256_values if str(value).strip()}
return self.catalog.query_files(root=self.directory, sha256=values)
[docs]
def find_file_by_id(self, file_id: int) -> dict[str, Any] | None:
"""Return a current location for one attachment identifier."""
records = self.catalog.query_files(
root=self.directory, attachment_id=int(file_id)
)
return records[0] if records else None
[docs]
def find_location_by_id(self, location_id: int) -> dict[str, Any] | None:
"""Return a current location for one location identifier."""
records = self.catalog.query_files(
root=self.directory, location_id=int(location_id)
)
return records[0] if records else None
[docs]
def find_file_by_relative_path(
self, root: Path, relative_path: str
) -> dict[str, Any] | None:
"""Return the current location at a root-relative path."""
return self.catalog.get_file(root, relative_path)
[docs]
def find_files_by_bibcode(self, root: Path, bibcode: str) -> list[dict[str, Any]]:
"""Return unchanged current locations carrying one Bibcode."""
return [
record
for record in self.catalog.query_files(root=root)
if str(record.get("bibcode") or "").strip() == bibcode
and not bool(
record.get(METADATA_STALE_KEY)
or (
isinstance(record.get("metadata"), dict)
and record["metadata"].get(METADATA_STALE_KEY)
)
)
and (
record.get("status") == "resolved"
or stored_bibcode_matches_file(record)
)
]
[docs]
def find_file_by_current_path(
self, root: Path, current_path: str
) -> dict[str, Any] | None:
"""Return the current location at ``current_path``."""
return self.catalog.get_file(root, current_path)
[docs]
def update_file_status(self, root: Path, relative_path: str, status: str) -> bool:
"""Update one attachment's local status."""
return self.catalog.update_file_status(root, relative_path, status)
[docs]
def record_rename(
self,
root: Path,
relative_path: str,
current_path: str,
new_relative_path: str | None = None,
) -> bool:
"""Record a location move while retaining attachment identity."""
return self.catalog.move_file(
root,
relative_path,
new_relative_path or current_path,
)
[docs]
def iter_files(
self,
status: str | None = None,
*,
include_missing: bool = False,
) -> Iterator[dict[str, Any]]:
"""Yield locations in this root, optionally including missing history."""
yield from self.catalog.query_files(
root=self.directory,
status=status,
include_missing=include_missing,
)
[docs]
def count_files_matching(self, **filters: Any) -> int:
"""Count current locations matching supported catalog filters.
Counts in SQL when the filters allow it. The `source` filter reads
metadata JSON. Identification groups also read current identity evidence.
`duplicates_only` compares digests and paper records across the
filtered rows, so those still load the records.
"""
sql_filters = dict(filters)
source = sql_filters.pop("source", None)
duplicates_only = sql_filters.pop("duplicates_only", False)
identification = sql_filters.pop("identification", None)
digest = sql_filters.get("sha256")
if (
source is None
and identification is None
and not duplicates_only
and (digest is None or isinstance(digest, str))
):
return self.catalog.count_files(root=self.directory, **sql_filters)
return len(self.query_records(**filters))
[docs]
def query_files(
self,
*,
order_by: str = "relative_path",
descending: bool = False,
secondary_order_by: str | None = None,
tertiary_order_by: str | None = None,
limit: int | None = None,
offset: int = 0,
**filters: Any,
) -> list[dict[str, Any]]:
"""Return sorted and paged current locations for this root."""
if order_by not in SORTABLE_COLUMNS:
raise ValueError(f"Unsortable column: {order_by!r}")
if limit is not None and limit < 0:
raise ValueError("limit must be non-negative or None")
if offset < 0:
raise ValueError("offset must be non-negative")
order_keys = [
key
for key in (order_by, secondary_order_by, tertiary_order_by)
if key is not None
]
invalid = next(
(key for key in order_keys if key not in SORTABLE_COLUMNS),
None,
)
if invalid is not None:
raise ValueError(f"Unsortable column: {invalid!r}")
duplicates_only = bool(filters.get("duplicates_only"))
records = self.query_records(**filters)
self.sort_records(
records,
order_by=order_by,
descending=descending,
secondary_order_by=secondary_order_by,
tertiary_order_by=tertiary_order_by,
)
if duplicates_only:
order_duplicate_groups(records)
if limit is None:
return records[offset:]
return records[offset : offset + int(limit)]
[docs]
def sort_records(
self,
records: list[dict[str, Any]],
*,
order_by: str = "relative_path",
descending: bool = False,
secondary_order_by: str | None = None,
tertiary_order_by: str | None = None,
) -> None:
"""Sort `records` in place with the `query_files` order rules.
Empty values sort last for each key, in both directions.
"""
order_keys = [
key
for key in (order_by, secondary_order_by, tertiary_order_by)
if key is not None
]
invalid = next((key for key in order_keys if key not in SORTABLE_COLUMNS), None)
if invalid is not None:
raise ValueError(f"Unsortable column: {invalid!r}")
for key in reversed(order_keys):
records.sort(
key=lambda record, sort_key=key: self._sort_value(record, sort_key)[1],
reverse=descending,
)
records.sort(
key=lambda record, sort_key=key: self._sort_value(record, sort_key)[0]
)
[docs]
def change_token(self) -> tuple[str, tuple[int, ...]]:
"""Return this root and the catalog change token."""
return str(self.directory), self.catalog.change_token()
[docs]
def delete_files_by_ids(self, file_ids: list[int]) -> int:
"""Delete selected current locations and orphaned attachments."""
return self.catalog.delete_locations(self.directory, file_ids)
[docs]
def upsert_ads_document(self, document: Mapping[str, Any], *, source: str) -> None:
"""Cache one ADS document in the central catalog."""
self.catalog.upsert_ads_document(document, source=source)
[docs]
def upsert_ads_documents(
self, documents: list[Mapping[str, Any]], *, source: str
) -> None:
"""Cache ADS documents atomically in the central catalog."""
self.catalog.upsert_ads_documents(documents, source=source)
[docs]
def find_ads_document_by_bibcode(self, bibcode: str) -> dict[str, Any] | None:
"""Return a cached ADS document by Bibcode."""
return self.catalog.find_ads_document_by_bibcode(bibcode)
[docs]
def set_ads_document_metrics(
self,
bibcode: str,
metrics: Mapping[str, int | None],
) -> bool:
"""Store cached citation metrics."""
return self.catalog.set_ads_document_metrics(bibcode, metrics)
[docs]
def find_ads_documents_by_bibcodes(
self, bibcodes: list[str]
) -> dict[str, dict[str, Any]]:
"""Return cached documents by Bibcode."""
return self.catalog.find_ads_documents_by_bibcodes(bibcodes)
[docs]
def find_ads_documents_by_doi(self, doi: str) -> list[dict[str, Any]]:
"""Return cached documents by normalized DOI."""
return self.catalog.find_ads_documents_by_doi(doi)
[docs]
def find_ads_documents_by_dois(
self, dois: list[str]
) -> dict[str, list[dict[str, Any]]]:
"""Return cached documents keyed by normalized DOI."""
return self.catalog.find_ads_documents_by_dois(dois)
[docs]
def export_payload(self) -> dict[str, Any]:
"""Return this root as a sidecar-shaped export."""
payload: dict[str, Any] = {"version": 2, "files": {}}
profile = self.get_meta("identify_profile")
if profile:
payload["identify_profile"] = json.loads(profile)
settings = self.catalog.get_meta(f"root:{self.directory}:identify_settings")
if settings:
payload["identify_settings"] = json.loads(settings)
rename = self.catalog.get_meta(f"root:{self.directory}:rename_settings")
if rename:
payload["rename_settings"] = json.loads(rename)
for record in self.iter_files():
payload["files"][record["relative_path"]] = {
**record.get("metadata", {}),
"current_path": record["current_path"],
"sha256": record["sha256"],
"status": record["status"],
}
return payload
[docs]
def save(self) -> Path:
"""Commit pending catalog writes and return the central path."""
self._connection.commit()
return self.path
[docs]
def close(self) -> None:
"""Close the central catalog connection."""
self.catalog.close()
[docs]
def query_records(self, **filters: Any) -> list[dict[str, Any]]:
"""Return filtered current locations in catalog order, without sorting.
Catalog order is ``relative_path`` and then location identifier.
`query_files` sorts this list.
"""
source = filters.pop("source", None)
duplicates_only = bool(filters.pop("duplicates_only", False))
identification = filters.pop("identification", None)
if identification is not None and identification not in IDENTIFICATION_GROUPS:
raise ValueError(f"Unknown identification group: {identification!r}")
records = self.catalog.query_files(root=self.directory, **filters)
if identification is not None:
records = [
record
for record in records
if identification_group(record) == identification
]
if source is not None:
records = [
record
for record in records
if str(record.get("source") or "") == str(source)
]
if duplicates_only:
records = mark_duplicates(records)
return records
@staticmethod
def _sort_value(record: Mapping[str, Any], key: str) -> tuple[bool, str | int]:
if key in STORED_COUNT_KEYS:
count = stored_count(record, key)
return (True, 0) if count is None else (False, count)
if key in NUMERIC_SORT_COLUMNS:
try:
return False, int(record[key])
except (KeyError, TypeError, ValueError):
return True, 0
if key == "status":
resolved = (
bool(record.get("bibcode")) and record.get("status") == "resolved"
)
return resolved, str(record.get("current_path") or "")
if key == "filename":
value = record.get("current_path")
elif key == "relative_path":
value = record.get("relative_path")
elif key == "sha256":
value = record.get("sha256")
elif key == "arxiv":
value = arxiv_id(record)
else:
value = record.get(key)
if value in (None, "", []):
value = (record.get("metadata") or {}).get(key)
if isinstance(value, (list, tuple)):
value = value[0] if value else ""
return value in (None, "", []), str(value or "")
[docs]
class CentralPdfRegistryState:
"""State adapter exposing current catalog locations to CLI workflows."""
def __init__(self, registry: CentralPdfRegistry) -> None:
self.registry = registry
self.directory = registry.directory
self.path = registry.path
@staticmethod
def _record_view(record: Mapping[str, Any]) -> dict[str, Any]:
metadata = copy.deepcopy(record.get("metadata") or {})
return {
**metadata,
"current_filename": record["current_path"],
"sha256": record["sha256"],
"status": record["status"],
"document_id": record.get("document_id"),
}
[docs]
@property
def files(self) -> dict[str, dict[str, Any]]:
"""Return current location metadata keyed by relative path."""
return {
str(record["relative_path"]): self._record_view(record)
for record in self.registry.iter_files()
}
[docs]
@property
def identify_profile(self) -> dict[str, Any] | None:
"""Return the saved identify profile."""
raw = self.registry.get_meta("identify_profile")
return json.loads(raw) if raw else None
[docs]
@property
def identify_settings(self) -> IdentifySettings | None:
"""Return root-specific reusable Identify settings."""
raw = self.registry.catalog.get_meta(f"root:{self.directory}:identify_settings")
if not raw:
return None
try:
payload = json.loads(raw)
except json.JSONDecodeError:
return None
return IdentifySettings.from_mapping(payload)
[docs]
@property
def rename_settings(self) -> RenameSettings | None:
"""Return root-specific reusable Rename settings."""
from .rename_settings import RenameSettings
raw = self.registry.catalog.get_meta(f"root:{self.directory}:rename_settings")
if not raw:
return None
try:
payload = json.loads(raw)
except json.JSONDecodeError:
return None
return RenameSettings.from_mapping(payload)
[docs]
def get_file(self, relative_path: str) -> dict[str, Any] | None:
"""Return one current location in sidecar-shaped form."""
record = self.registry.find_file_by_relative_path(self.directory, relative_path)
return self._record_view(record) if record is not None else None
[docs]
def find_by_sha256(self, sha256: str) -> tuple[str, dict[str, Any]] | None:
"""Find a current location by digest."""
record = self.registry.find_file_by_sha256(sha256)
if record is None:
return None
return str(record["relative_path"]), self._record_view(record)
[docs]
def find_by_current_filename(
self, filename: str
) -> tuple[str, dict[str, Any]] | None:
"""Find a current location by root-relative path."""
record = self.registry.find_file_by_current_path(self.directory, filename)
if record is None:
return None
return str(record["relative_path"]), self._record_view(record)
[docs]
def upsert_file(
self, relative_path: str, metadata: Mapping[str, Any]
) -> dict[str, Any]:
"""Register one current location through the central catalog."""
record = dict(metadata)
current = str(record.get("current_filename", relative_path))
path = self.directory / current
try:
stat = path.stat()
size_bytes, mtime_ns = stat.st_size, stat.st_mtime_ns
device, inode = stat.st_dev, stat.st_ino
except OSError:
size_bytes = int(record.get("size_bytes", 0) or 0)
mtime_ns = int(record.get("mtime_ns", 0) or 0)
device = inode = None
status = str(
record.get("status")
or ("resolved" if record.get("bibcode") else "unresolved")
)
proof = record.get("_adsfolio_identity")
if (
isinstance(proof, Mapping)
and proof.get("status") == "verified"
and path.is_file()
):
replacement = self.registry.catalog.register_verified_replacement(
self.directory, current, record
)
if replacement is not None:
stored = self.registry.catalog.get_file(self.directory, current)
if stored is None:
raise RuntimeError(f"Replacement location disappeared: {current}")
return self._record_view(stored)
self.registry.catalog.observe_file(
self.directory,
current,
sha256=str(record.get("sha256", "") or "") or None,
size_bytes=size_bytes,
mtime_ns=mtime_ns,
device=device,
inode=inode,
metadata=record,
status=status,
)
stored = self.registry.find_file_by_current_path(self.directory, current)
if stored is None:
raise RuntimeError(f"Catalog location disappeared after upsert: {current}")
return self._record_view(stored)
[docs]
def set_current_filename(self, relative_path: str, current_filename: str) -> bool:
"""Record a move between current root-relative locations."""
record = self.registry.find_file_by_relative_path(self.directory, relative_path)
if record is None:
return False
return self.registry.record_rename(
self.directory,
str(record["relative_path"]),
current_filename,
new_relative_path=current_filename,
)
[docs]
def set_identify_profile(self, profile: Mapping[str, Any] | None) -> bool:
"""Set identify profile metadata and report whether it changed."""
previous = self.identify_profile
normalized = dict(profile) if profile is not None else None
if previous == normalized:
return False
self.registry.set_meta(
"identify_profile",
"" if normalized is None else json.dumps(normalized, sort_keys=True),
)
return True
[docs]
def set_identify_settings(self, settings: IdentifySettings | None) -> bool:
"""Set root-specific reusable Identify settings."""
if self.identify_settings == settings:
return False
key = f"root:{self.directory}:identify_settings"
if settings is None:
self.registry.catalog.delete_meta(key)
else:
self.registry.catalog.set_meta(
key,
json.dumps(settings.to_mapping(), sort_keys=True),
)
return True
[docs]
def set_rename_settings(self, settings: RenameSettings | None) -> bool:
"""Set root-specific reusable Rename settings."""
if self.rename_settings == settings:
return False
key = f"root:{self.directory}:rename_settings"
if settings is None:
self.registry.catalog.delete_meta(key)
else:
self.registry.catalog.set_meta(
key,
json.dumps(settings.to_mapping(), sort_keys=True),
)
return True
[docs]
def save(self) -> Path:
"""Commit pending catalog writes."""
return self.registry.save()
[docs]
def duplicates_by_sha256(self) -> dict[str, list[str]]:
"""Return duplicate current locations grouped by digest."""
duplicates: dict[str, list[str]] = {}
for record in self.registry.iter_files():
digest = str(record.get("sha256") or "")
if digest:
duplicates.setdefault(digest, []).append(record["relative_path"])
return {digest: paths for digest, paths in duplicates.items() if len(paths) > 1}
[docs]
def close(self) -> None:
"""Close the central catalog."""
self.registry.close()
PdfRegistry = CentralPdfRegistry
PdfRegistryState = CentralPdfRegistryState
[docs]
def migrate_sidecar(json_path: Path, registry: CentralPdfRegistry) -> int:
"""Import legacy root data without modifying its source files."""
return registry.catalog.migrate_legacy_root(json_path.parent).imported
[docs]
def open_registry_state(directory: Path) -> CentralPdfRegistryState:
"""Open a central catalog state view for one library root."""
return CentralPdfRegistryState(CentralPdfRegistry(directory))