#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Claim lifecycle mutations — remove / supersede (single + by-prefix)."""
from __future__ import annotations
import os
from datetime import datetime
from pathlib import Path
from scitex_dev.store import ANY_REVISION
from .._db import get_db
from ._model import _file_path_matches_prefix, _resolve_claim
from ._store import _open_store
def _auto_export(context: str) -> None:
"""Re-emit claims.json after a mutation (best-effort, never fatal)."""
if os.environ.get("SCITEX_CLEW_AUTO_EXPORT_CLAIMS", "1") == "0":
return
try:
from ._export import export_claims_json
export_claims_json()
except Exception as exc: # noqa: BLE001
import warnings as _w
_w.warn(
f"scitex_clew auto-export of claims.json failed after {context}: {exc!r}",
RuntimeWarning,
stacklevel=3,
)
[docs]
def remove_claim(claim_id_or_location: str) -> bool:
"""Hard-delete a claim from the database.
Permanently removes the claim row identified by ``claim_id_or_location``
(a claim_id string, a location like ``"paper.tex:L42"``, or a bare file
path — resolved via the same logic as :func:`verify_claim`).
After deletion :func:`export_claims_json` is called so the JSON
artifact stays in sync with the DB.
Parameters
----------
claim_id_or_location : str
Claim identifier. Resolution order:
1. Exact ``claim_id`` match.
2. Location string ``"file.tex:L42"``.
3. File path only (first row).
Returns
-------
bool
``True`` if a row was deleted; ``False`` if nothing matched.
"""
db = get_db()
claim = _resolve_claim(claim_id_or_location, db)
if claim is None:
return False
# A Store never deletes a row — hide() sets the ``hidden`` flag instead
# (the ONLY removal it offers). Every read in this package defaults to
# ``include_hidden=False``, so a hidden claim reads exactly like the old
# hard-deleted one everywhere except a caller that deliberately asks for
# the hidden view.
store = _open_store()
try:
store.hide({"claim_id": claim.claim_id}, expected_revision=ANY_REVISION)
finally:
store.close()
_auto_export("remove_claim")
return True
def remove_claims_by_prefix(file_path_prefix: str) -> int:
"""Hard-delete all claims whose file_path starts with ``file_path_prefix``.
Parameters
----------
file_path_prefix : str
Path prefix (resolved). All claims under this root are deleted.
Returns
-------
int
Number of rows deleted.
"""
resolved_prefix = str(Path(file_path_prefix).resolve())
if not resolved_prefix.endswith("/"):
resolved_prefix = resolved_prefix + "/"
store = _open_store()
try:
deleted = 0
for row in store.rows():
if _file_path_matches_prefix(row.values["file_path"], resolved_prefix):
store.hide(
{"claim_id": row.values["claim_id"]}, expected_revision=ANY_REVISION
)
deleted += 1
finally:
store.close()
_auto_export("remove_claims_by_prefix")
return deleted
[docs]
def supersede_claim(claim_id_or_location: str) -> bool:
"""Soft-retire a claim by setting its status to ``"superseded"``.
The row is kept in the database (audit trail) but excluded from the
default :func:`list_claims` view (``include_superseded=False`` is the
default), from :func:`verify_all_claims`, and from the default
:func:`export_claims_json` output.
This allows a user to retire stale/dead claims so that ``clew verify``
can reach exit 0 without deleting the historical record.
Parameters
----------
claim_id_or_location : str
Claim identifier resolved the same way as :func:`remove_claim`.
Returns
-------
bool
``True`` if the claim existed and was updated; ``False`` if nothing
matched.
"""
db = get_db()
claim = _resolve_claim(claim_id_or_location, db)
if claim is None:
return False
store = _open_store()
try:
store.put(
{
"claim_id": claim.claim_id,
"status": "superseded",
"verified_at": datetime.now().isoformat(),
},
expected_revision=ANY_REVISION,
)
finally:
store.close()
_auto_export("supersede_claim")
return True
def supersede_claims_by_prefix(file_path_prefix: str) -> int:
"""Soft-retire all claims whose file_path starts with ``file_path_prefix``.
Parameters
----------
file_path_prefix : str
Path prefix (resolved).
Returns
-------
int
Number of rows updated to status ``"superseded"``.
"""
resolved_prefix = str(Path(file_path_prefix).resolve())
if not resolved_prefix.endswith("/"):
resolved_prefix = resolved_prefix + "/"
store = _open_store()
try:
now = datetime.now().isoformat()
updated = 0
for row in store.rows():
if _file_path_matches_prefix(row.values["file_path"], resolved_prefix):
store.put(
{
"claim_id": row.values["claim_id"],
"status": "superseded",
"verified_at": now,
},
expected_revision=ANY_REVISION,
)
updated += 1
finally:
store.close()
_auto_export("supersede_claims_by_prefix")
return updated
# EOF