Skip to content

zettelkasten.dashboard.backend.routes.citations

zettelkasten.dashboard.backend.routes.citations

Citation detail, source-citation, promote/unlink, and search routes.

Split out of the former flat routes.py; behaviour is unchanged.

get_citation

get_citation(citation_id: str) -> dict

Bibliographic detail for a single citation (the global _citations/ store).

A pure citation is a work with no writable source folder, so it has no note detail to show — this surfaces its stored bibliographic metadata for the dashboard right pane. Federated-repo citation detail is out of scope; a non-local id simply 404s.

Source code in zettelkasten/dashboard/backend/routes/citations.py
@router.get("/citations/{citation_id}")
def get_citation(citation_id: str) -> dict:
    """Bibliographic detail for a single citation (the global ``_citations/`` store).

    A pure citation is a work with no writable source folder, so it has no note
    detail to show — this surfaces its stored bibliographic metadata for the
    dashboard right pane. Federated-repo citation detail is out of scope; a
    non-local id simply 404s.
    """
    citations = load_citations(graphs_dir=GRAPHS_DIR)
    cdata = citations.get(citation_id)
    if cdata is None:
        raise HTTPException(status_code=404, detail=f"Citation '{citation_id}' not found.")
    return _hydrate_citation(citation_id, cdata)

get_source_citations

get_source_citations(name: str) -> dict

The citation works connected to a source via its _meta.yaml.

Surfaces the references (works this source cites — backward) and cited_by (works that cite this source — forward) id lists, hydrated from the global _citations/ store, so the source pane can show an expandable list instead of citations floating loose in _citations/. Unresolvable ids (a stub that was promoted or deleted) are reported separately rather than silently dropped. Promoted ids become navigable source links.

Source code in zettelkasten/dashboard/backend/routes/citations.py
@router.get("/graphs/{name}/citations")
def get_source_citations(name: str) -> dict:
    """The citation works connected to a source via its ``_meta.yaml``.

    Surfaces the ``references`` (works this source cites — backward) and
    ``cited_by`` (works that cite this source — forward) id lists, hydrated from
    the global ``_citations/`` store, so the source pane can show an expandable
    list instead of citations floating loose in ``_citations/``. Unresolvable
    ids (a stub that was promoted or deleted) are reported separately rather
    than silently dropped. Promoted ids become navigable source links.
    """
    meta = load_source_meta(name, graphs_dir=GRAPHS_DIR)
    if not meta and not (GRAPHS_DIR / name).is_dir():
        raise HTTPException(status_code=404, detail=f"Source '{name}' not found.")

    citations = load_citations(graphs_dir=GRAPHS_DIR)

    def _resolve(ids: object) -> tuple[list[dict], list[str]]:
        resolved: list[dict] = []
        missing: list[str] = []
        seen: set[str] = set()
        for cid in (ids if isinstance(ids, list) else []):
            if not isinstance(cid, str) or cid in seen:
                continue
            seen.add(cid)
            cdata = citations.get(cid)
            if cdata is not None:
                resolved.append(_hydrate_citation(cid, cdata))
            elif (GRAPHS_DIR / cid / "_meta.yaml").is_file():
                # The stub was promoted to a full source — link to it directly.
                smeta = load_source_meta(cid, graphs_dir=GRAPHS_DIR)
                resolved.append({
                    "id": cid,
                    "title": _cit_as_str(smeta.get("title", "")),
                    "authors": _cit_as_list(smeta.get("authors", [])),
                    "year": _cit_as_int(smeta.get("year", 0)),
                    "venue": _cit_as_str(smeta.get("venue", "")),
                    "type": _cit_as_str(smeta.get("doc_type", "")),
                    "doi": _cit_as_str(smeta.get("doi", "")),
                    "openalex_id": "",
                    "cited_by_count": _cit_as_int(smeta.get("cited_by_count", 0)),
                    "counts_by_year": [],
                    "abstract": _cit_as_str(smeta.get("abstract", "")),
                    "promoted": True,
                })
            else:
                missing.append(cid)
        resolved.sort(key=lambda c: (c.get("cited_by_count") or 0), reverse=True)
        return resolved, missing

    references, ref_missing = _resolve(meta.get("references"))
    cited_by, cb_missing = _resolve(meta.get("cited_by"))
    return {
        "source": name,
        "references": references,
        "cited_by": cited_by,
        "missing": sorted(set(ref_missing + cb_missing)),
    }

promote_citation_to_source

promote_citation_to_source(citation_id: str, req: PromoteCitationRequest | None = None) -> dict

Promote an OpenAlex citation into an owned source (a graph), gated on Zotero.

The write-side action behind the per-citation "promote to a graph" button: it confirms the work already lives in the user's Zotero library (matching by DOI/title), then pulls in the PDF, ingests it, promotes the citation stub into an owned source folder, and wires the full text — preserving the citation edges that already point at the stub. When the work isn't in Zotero yet, NOTHING is mutated and the result asks the user to add it to Zotero first (status="not_in_zotero").

Promotion is a WRITE that also removes the citation stub, so it is blocked with 403 under ZK_DISABLE_WRITE / ZK_DISABLE_DELETE (mirroring the review-action barrier). An unexpected raise is returned as a clean structured failed result rather than a 500. status is one of acquired | already_owned | not_in_zotero | zotero_unconfigured | failed.

Source code in zettelkasten/dashboard/backend/routes/citations.py
@router.post("/citations/{citation_id}/promote")
def promote_citation_to_source(
    citation_id: str, req: PromoteCitationRequest | None = None
) -> dict:
    """Promote an OpenAlex citation into an owned source (a graph), gated on Zotero.

    The write-side action behind the per-citation "promote to a graph" button:
    it confirms the work already lives in the user's Zotero library (matching by
    DOI/title), then pulls in the PDF, ingests it, promotes the citation stub
    into an owned source folder, and wires the full text — preserving the
    citation edges that already point at the stub. When the work isn't in Zotero
    yet, NOTHING is mutated and the result asks the user to add it to Zotero first
    (``status="not_in_zotero"``).

    Promotion is a WRITE that also removes the citation stub, so it is blocked
    with 403 under ``ZK_DISABLE_WRITE`` / ``ZK_DISABLE_DELETE`` (mirroring the
    review-action barrier). An unexpected raise is returned as a clean structured
    ``failed`` result rather than a 500. ``status`` is one of ``acquired`` |
    ``already_owned`` | ``not_in_zotero`` | ``zotero_unconfigured`` | ``failed``.
    """
    import zettelkasten.server as zk_server

    if zk_server._WRITE_DISABLED:
        raise HTTPException(
            status_code=403,
            detail="Graph writes are disabled (ZK_DISABLE_WRITE); this action is unavailable.",
        )
    if zk_server._DELETE_DISABLED:
        raise HTTPException(
            status_code=403,
            detail="Graph deletes are disabled (ZK_DISABLE_DELETE); promotion removes the citation stub.",
        )

    zotero_key = (req.zotero_key if req else "") or ""
    try:
        result = json.loads(
            zk_server.promote_citation_to_graph(citation_id=citation_id, zotero_key=zotero_key)
        )
    except Exception as exc:
        return {
            "status": "failed",
            "target": citation_id,
            "error": str(exc),
            "type": exc.__class__.__name__,
            "message": f"Failed to promote '{citation_id}': {exc}",
        }

    # A successful promote changes a source's ownership/coverage, which the papers
    # and claims read-views derive from — drop their caches so the next fetch
    # reflects the new corpus (mirrors the review-action and coverage routes).
    if result.get("status") in ("acquired", "already_owned"):
        _papers_cache.clear()
        _claims_cache.clear()
    return result
unlink_citation_from_scope(citation_id: str, req: UnlinkCitationRequest) -> dict

Remove a "proposed" citation's links from the graph — scoped or globally.

The write-side action behind the per-citation "Unlink / Remove" button in the Papers view. Two tiers (selected by remove_global):

  • Unlink from this view (default): drops the graph="_citations" note links that point at the citation within the scope (a project's sources, or a single source graph) and clears it from those sources' references/cited_by lists. The global citation entity is KEPT. Classified as a WRITE — refused under ZK_DISABLE_WRITE only.
  • Remove everywhere (remove_global=true): also scrubs every other graph and deletes _citations/<id>.yaml. A DELETE, so additionally refused under ZK_DISABLE_DELETE (mirroring the promote route).

LOCAL-ONLY: a federated/read-only scope is rejected. Clears the papers/claims caches so the removed work disappears from the scored views.

Source code in zettelkasten/dashboard/backend/routes/citations.py
@router.post("/citations/{citation_id}/unlink")
def unlink_citation_from_scope(citation_id: str, req: UnlinkCitationRequest) -> dict:
    """Remove a "proposed" citation's links from the graph — scoped or globally.

    The write-side action behind the per-citation "Unlink / Remove" button in the
    Papers view. Two tiers (selected by ``remove_global``):

    * **Unlink from this view** (default): drops the ``graph="_citations"`` note
      links that point at the citation within the scope (a project's sources, or a
      single source graph) and clears it from those sources' references/cited_by
      lists. The global citation entity is KEPT. Classified as a WRITE — refused
      under ``ZK_DISABLE_WRITE`` only.
    * **Remove everywhere** (``remove_global=true``): also scrubs every other graph
      and deletes ``_citations/<id>.yaml``. A DELETE, so additionally refused under
      ``ZK_DISABLE_DELETE`` (mirroring the promote route).

    LOCAL-ONLY: a federated/read-only scope is rejected. Clears the papers/claims
    caches so the removed work disappears from the scored views.
    """
    import zettelkasten.server as zk_server

    if zk_server._WRITE_DISABLED:
        raise HTTPException(
            status_code=403,
            detail="Graph writes are disabled (ZK_DISABLE_WRITE); this action is unavailable.",
        )
    if req.remove_global and zk_server._DELETE_DISABLED:
        raise HTTPException(
            status_code=403,
            detail="Graph deletes are disabled (ZK_DISABLE_DELETE); removing the citation everywhere deletes its entity.",
        )

    # Resolve the in-scope source graphs and reject a federated (read-only) scope.
    if req.project:
        _load_project_or_404(req.project)
        local, repo = _resolve_ref(req.project)
        if repo is not None:
            raise HTTPException(status_code=400, detail="Citation unlink is local-repo only.")
        project_data = _load_project_or_404(req.project)
        sources = list(project_data.get("sources", []))
        # Cross-graph notes can surface a project's citation in its graph/Papers
        # view, so include _cross in the scope it is removed from.
        if (GRAPHS_DIR / "_cross").is_dir():
            sources.append("_cross")
    elif req.graph:
        local, repo = _resolve_ref(req.graph)
        if repo is not None:
            raise HTTPException(status_code=400, detail="Citation unlink is local-repo only.")
        sources = [req.graph]
    elif req.remove_global:
        sources = []  # global remove scans every graph itself
    else:
        raise HTTPException(status_code=400, detail="Provide a project or graph scope, or set remove_global.")

    result = json.loads(zk_server.unlink_citation(
        citation_id=citation_id,
        sources=sources,
        remove_global=req.remove_global,
        dry_run=req.dry_run,
    ))
    if isinstance(result, dict) and result.get("error"):
        raise HTTPException(status_code=400, detail=result["error"])
    if not req.dry_run:
        _papers_cache.clear()
        _claims_cache.clear()
    return result