path: root/bookmarks/bookmarks.py

#!/usr/bin/python3
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.

"""Firefox bookmarks as JSON, grouped by folder in Firefox's own order, each
with its favicon extracted to a cache file and the result of the last link
check.

    bookmarks.py           print the list
    bookmarks.py --links   check every http(s) bookmark, write the results
    bookmarks.py --check   run the self-check

The databases are read from a copy: Firefox holds them in exclusive lock
mode while it runs, so even a read-only open fails, and this only reads. A
write would need Firefox closed and is not attempted.
"""

import concurrent.futures
import hashlib
import json
import os
import shutil
import socket
import sqlite3
import ssl
import sys
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path

FIREFOX = Path("~/.mozilla/firefox").expanduser()
CACHE = Path(os.environ.get("XDG_CACHE_HOME", "~/.cache")).expanduser() / "keystone/bookmarks"
LINKS = CACHE / "links.json"

# Root folders by guid, in the order they are shown. Tags are left out: they
# are not bookmarks, and each one repeats a bookmark found elsewhere.
ROOTS = {"toolbar_____": "Toolbar", "menu________": "Menu",
         "unfiled_____": "Other Bookmarks", "mobile______": "Mobile"}

MAGIC = [(b"\x89PNG", "png"), (b"<svg", "svg"), (b"<?xml", "svg"),
         (b"\x00\x00\x01\x00", "ico"), (b"GIF8", "gif"), (b"\xff\xd8", "jpg"),
         (b"RIFF", "webp")]


def running(prof):
    """Whether a live Firefox holds this profile. Its `lock` symlink points at
    "<ip>:+<pid>" and outlives a crash, so the pid is checked too."""
    try:
        pid = os.readlink(prof / "lock").rpartition("+")[2]
        return "firefox" in Path(f"/proc/{pid}/comm").read_text()
    except OSError:
        return False


def profile():
    """BOOKMARKS_PROFILE, else the profile a running Firefox holds, else the
    one with the largest places.sqlite. Not installs.ini's Default, which
    names whichever profile the last plain launch picked, here the near-empty
    guest one; nor the last-modified places.sqlite, which a stray launch of
    that guest profile makes newest."""
    if os.environ.get("BOOKMARKS_PROFILE"):
        return Path(os.environ["BOOKMARKS_PROFILE"]).expanduser()
    places = sorted(FIREFOX.glob("*/places.sqlite"), key=lambda p: p.stat().st_size)
    if not places:
        sys.exit("no Firefox profile with a places.sqlite")
    live = [p for p in places if running(p.parent)]
    # ponytail: two profiles running at once, the bigger wins; a toggle in
    # the overlay if both are ever wanted.
    return (live or places)[-1].parent


_copies = tempfile.TemporaryDirectory()


def connect(path):
    """A copy of the database with its write-ahead log. Firefox writes every
    change to the -wal first and folds it in at a checkpoint, maybe minutes
    later. `immutable=1` on the original skipped the log, so a bookmark
    deleted in Firefox stayed listed until then. Copied into a directory of
    this process's own, so the list and a link check never share one."""
    # ponytail: a copy taken mid-checkpoint could be torn; a reopen fixes it.
    dst = Path(_copies.name) / path.name
    for suffix in ("", "-wal"):
        if Path(f"{path}{suffix}").exists():
            shutil.copyfile(f"{path}{suffix}", f"{dst}{suffix}")
    return sqlite3.connect(dst)


def tree(rows):
    """rows: (id, parent, type, title, position, guid, url, added, visits).
    Returns [(folder path, [(title, url, added, visits)])], one group per folder holding bookmarks:
    the folder's own bookmarks first, then its subfolders, each in Firefox's
    order. A folder's bookmarks and subfolders interleave in Firefox; grouping
    them is what makes the list readable."""
    kids = {}
    for r in sorted(rows, key=lambda r: r[4]):
        kids.setdefault(r[1], []).append(r)
    groups = []

    def walk(fid, path):
        links = [(r[3] or r[6], r[6], r[7], r[8]) for r in kids.get(fid, []) if r[2] == 1 and r[6]]
        if links:
            groups.append((" / ".join(path), links))
        for r in kids.get(fid, []):
            if r[2] == 2:
                walk(r[0], path + [r[3] or "(untitled)"])

    by_guid = {r[5]: r for r in rows}
    for guid, name in ROOTS.items():
        if guid in by_guid:
            walk(by_guid[guid][0], [name])
    return groups


def pick_icon(icons):
    """icons: [(width, data)]. The smallest at least 32px wide, else the
    largest. An SVG has width 65535, so it wins only when nothing is 32-64."""
    big = [i for i in icons if i[0] >= 32]
    return min(big)[1] if big else max(icons)[1]


def icon_file(data):
    """Write data to the cache under its hash, return the path, or "" for a
    format Qt cannot show. Content-addressed, so a changed icon is a new file
    and nothing needs invalidating."""
    ext = next((e for m, e in MAGIC if data.startswith(m)), None)
    if not ext:
        return ""
    path = CACHE / "icons" / f"{hashlib.sha1(data).hexdigest()}.{ext}"
    if not path.exists():
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(data)
    return str(path)


def icons(prof, urls):
    """{url: icon path}: the page's own icon, else any icon Firefox holds for
    the same host, from another page or the site's root /favicon.ico (stored
    once per site, linked to no page). The host fallback, scheme ignored,
    found 52 more of 504 bookmarks; the rest have no icon in Firefox at all,
    which expires them."""
    db = connect(prof / "favicons.sqlite")
    by_page, by_host = {}, {}
    for page, width, data in db.execute(
            "select p.page_url, i.width, i.data from moz_pages_w_icons p "
            "join moz_icons_to_pages ip on ip.page_id = p.id "
            "join moz_icons i on i.id = ip.icon_id where i.data is not null"):
        by_page.setdefault(page, []).append((width, data))
        by_host.setdefault(host(page), []).append((width, data))
    for url, width, data in db.execute(
            "select icon_url, width, data from moz_icons where root = 1 and data is not null"):
        by_host.setdefault(host(url), []).append((width, data))
    db.close()
    out = {}
    for u in urls:
        found = by_page.get(u) or by_host.get(host(u))
        if found:
            out[u] = icon_file(pick_icon(found))
    return out


def host(url):
    return url.partition("://")[2].split("/", 1)[0]


def main():
    prof = profile()
    db = connect(prof / "places.sqlite")
    rows = db.execute(
        "select b.id, b.parent, b.type, b.title, b.position, b.guid, p.url, "
        "b.dateAdded / 1000000, coalesce(p.visit_count, 0) "
        "from moz_bookmarks b left join moz_places p on p.id = b.fk").fetchall()
    db.close()
    groups = tree(rows)
    found = icons(prof, [e[1] for _, entries in groups for e in entries])
    links = json.loads(LINKS.read_text()) if LINKS.exists() else {}
    items = [{"folder": folder, "title": title, "url": url,
              "added": added, "visits": visits,
              "icon": found.get(url, ""),
              "state": links.get(url, {}).get("state", ""),
              "reason": links.get(url, {}).get("reason", "")}
             for folder, entries in groups for title, url, added, visits in entries]
    checked = time.strftime("%Y-%m-%d %H:%M", time.localtime(LINKS.stat().st_mtime)) \
        if LINKS.exists() else ""
    json.dump({"checked": checked, "items": items}, sys.stdout)


def classify(error=None, code=None):
    """(state, reason) for one request: "ok", "dead" for a page or host that
    is gone, "warn" for anything that may only be refusing a script.
    A 403 or 429 is usually a bot wall and a timeout may be a slow site, so
    neither is called dead."""
    if code is not None:
        if code < 400:
            return "ok", ""
        return ("dead" if code in (404, 410) else "warn"), f"HTTP {code}"
    reason = getattr(error, "reason", error)
    if isinstance(reason, socket.gaierror):
        # Only "name unknown" is an answer; EAI_AGAIN is the resolver
        # failing, which says nothing about the site.
        if reason.errno in (socket.EAI_NONAME, getattr(socket, "EAI_NODATA", None)):
            return "dead", "no such host"
        return "warn", "DNS failure"
    if isinstance(reason, ConnectionRefusedError):
        return "dead", "connection refused"
    if isinstance(reason, (TimeoutError, socket.timeout)):
        return "warn", "timeout"
    if isinstance(reason, ssl.SSLError):
        return "warn", "TLS error"
    return "warn", (getattr(reason, "strerror", None) or str(reason))[:60]


def probe(url):
    """HEAD first; any HEAD error gets a GET, body unread, so a server that
    mishandles HEAD is not called dead for it."""
    for method in ("HEAD", "GET"):
        req = urllib.request.Request(url, method=method, headers={
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) Gecko/20100101 Firefox/140.0"})
        try:
            with urllib.request.urlopen(req, timeout=15) as r:
                return classify(code=r.status)
        except urllib.error.HTTPError as e:
            if method == "HEAD":
                continue
            return classify(code=e.code)
        except Exception as e:
            return classify(error=e)


def check_links():
    db = connect(profile() / "places.sqlite")
    urls = sorted({u for (u,) in db.execute(
        "select p.url from moz_bookmarks b join moz_places p on p.id = b.fk "
        "where b.type = 1 and (p.url like 'http://%' or p.url like 'https://%')")})
    db.close()
    # ponytail: 32 at once, one pass, no retry; a site down for a minute reads
    # dead until the next check.
    with concurrent.futures.ThreadPoolExecutor(32) as pool:
        results = dict(zip(urls, pool.map(probe, urls)))
    LINKS.parent.mkdir(parents=True, exist_ok=True)
    tmp = LINKS.with_suffix(".tmp")
    tmp.write_text(json.dumps({u: {"state": s, "reason": r} for u, (s, r) in results.items()}))
    tmp.replace(LINKS)
    dead = sum(s == "dead" for s, _ in results.values())
    print(f"{len(urls)} checked, {dead} dead")


def check():
    rows = [(1, 0, 2, "", 0, "root________", None, 0, 0),
            (2, 1, 2, "menu", 0, "menu________", None, 0, 0),
            (3, 1, 2, "toolbar", 1, "toolbar_____", None, 0, 0),
            (4, 1, 2, "tags", 2, "tags________", None, 0, 0),
            (10, 3, 1, "B", 1, "b", "https://b.example.org/", 0, 0),
            (11, 3, 2, "Dev", 0, "dev", None, 0, 0),
            (12, 11, 1, "", 0, "c", "https://c.example.org/x", 0, 0),
            (13, 3, 1, "A", 2, "a", "https://a.example.org/", 0, 0),
            (14, 3, 3, "", 3, "sep", None, 0, 0),
            (15, 4, 1, "tagged", 0, "t", "https://t.example.org/", 0, 0),
            (16, 2, 2, "Empty", 0, "e", None, 0, 0)]
    assert tree(rows) == [
        ("Toolbar", [("B", "https://b.example.org/", 0, 0), ("A", "https://a.example.org/", 0, 0)]),
        ("Toolbar / Dev", [("https://c.example.org/x", "https://c.example.org/x", 0, 0)]),
    ], tree(rows)

    assert pick_icon([(16, b"s"), (32, b"m"), (64, b"l"), (65535, b"v")]) == b"m"
    assert pick_icon([(16, b"s"), (24, b"m")]) == b"m"
    assert host("https://a.example.org/x/y?z") == "a.example.org"

    assert classify(code=200) == ("ok", "")
    assert classify(code=404) == ("dead", "HTTP 404")
    assert classify(code=403) == ("warn", "HTTP 403")
    assert classify(error=urllib.error.URLError(socket.gaierror(-2, "x"))) == ("dead", "no such host")
    assert classify(error=urllib.error.URLError(socket.gaierror(socket.EAI_AGAIN, "x"))) == ("warn", "DNS failure")
    assert classify(error=urllib.error.URLError(ConnectionRefusedError())) == ("dead", "connection refused")
    assert classify(error=TimeoutError()) == ("warn", "timeout")
    assert classify(error=urllib.error.URLError(OSError(113, "No route to host"))) == ("warn", "No route to host")
    print("ok")


if __name__ == "__main__":
    if "--check" in sys.argv:
        check()
    elif "--links" in sys.argv:
        check_links()
    else:
        main()