Compare commits
4
Commits
db8887296b
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c28c955df8
|
||
|
|
b3f9539189
|
||
|
|
4df9c0116e
|
||
|
|
a7f9527344
|
No files matched your search
@@ -0,0 +1 @@
|
|||||||
|
*.pdf
|
||||||
@@ -0,0 +1,351 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
# /// script
|
||||||
|
# requires-python = ">=3.12"
|
||||||
|
# dependencies = [
|
||||||
|
# "pillow==12.3.0",
|
||||||
|
# "reportlab==5.0.1",
|
||||||
|
# ]
|
||||||
|
# ///
|
||||||
|
"""
|
||||||
|
Compose A4 print sheets for 1008×1792 themed images as cut-out bookmarks.
|
||||||
|
|
||||||
|
Design goals
|
||||||
|
------------
|
||||||
|
- Never resample the source pixels (quality identical to the generated file).
|
||||||
|
- Physical size is therefore determined only by the chosen print DPI.
|
||||||
|
- Thin crop marks + optional rounded-corner cut guides are drawn in vector
|
||||||
|
so they stay sharp at any printer resolution.
|
||||||
|
- Two ready layouts:
|
||||||
|
* 2-up (~300 dpi) → 85.3 × 151.7 mm bookmarks, two per A4
|
||||||
|
* 4-up (~356 dpi) → 72.0 × 128.0 mm bookmarks, four per A4
|
||||||
|
|
||||||
|
Usage
|
||||||
|
-----
|
||||||
|
python make_bookmark_sheets.py # write empty templates
|
||||||
|
python make_bookmark_sheets.py img1.png img2.png ...
|
||||||
|
python make_bookmark_sheets.py --layout 4up *.png
|
||||||
|
|
||||||
|
Requires: pillow, reportlab
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from PIL import Image
|
||||||
|
from reportlab.lib.pagesizes import A4
|
||||||
|
from reportlab.lib.units import mm
|
||||||
|
from reportlab.pdfbase import pdfmetrics
|
||||||
|
from reportlab.pdfbase.ttfonts import TTFont
|
||||||
|
from reportlab.pdfgen import canvas
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Constants
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
SRC_W, SRC_H = 1008, 1792
|
||||||
|
A4_W, A4_H = A4 # points
|
||||||
|
|
||||||
|
# Layout presets: name → (physical width mm, physical height mm, cols, rows)
|
||||||
|
# Height is derived from source aspect so there is no distortion.
|
||||||
|
ASPECT = SRC_H / SRC_W
|
||||||
|
|
||||||
|
LAYOUTS = {
|
||||||
|
"2up": {
|
||||||
|
"label": "2-up · ~300 dpi · larger bookmarks",
|
||||||
|
"width_mm": SRC_W / 300 * 25.4, # 85.344
|
||||||
|
"cols": 2,
|
||||||
|
"rows": 1,
|
||||||
|
"gap_mm": 10.0,
|
||||||
|
"margin_mm": 12.0,
|
||||||
|
"corner_r_mm": 4.0,
|
||||||
|
"dpi": 300.0,
|
||||||
|
},
|
||||||
|
"4up": {
|
||||||
|
"label": "4-up · ~365 dpi · four per sheet",
|
||||||
|
"width_mm": 70.0,
|
||||||
|
"cols": 2,
|
||||||
|
"rows": 2,
|
||||||
|
"gap_mm": 7.0,
|
||||||
|
"margin_mm": 10.0,
|
||||||
|
"corner_r_mm": 3.5,
|
||||||
|
"dpi": SRC_W / (70.0 / 25.4),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
MARK_COLOR = (0.15, 0.15, 0.15)
|
||||||
|
GUIDE_COLOR = (0.55, 0.15, 0.15)
|
||||||
|
NOTE_COLOR = (0.25, 0.25, 0.25)
|
||||||
|
|
||||||
|
|
||||||
|
# CachyOS is Arch-based: packages drop TTF/OTF under /usr/share/fonts/{TTF,OTF,…}
|
||||||
|
# rather than Debian's /usr/share/fonts/truetype/dejavu/.
|
||||||
|
FONT_ROOTS = [
|
||||||
|
Path("/usr/share/fonts"),
|
||||||
|
Path("/usr/local/share/fonts"),
|
||||||
|
Path("/usr/share/fonts/TTF"),
|
||||||
|
Path("/usr/share/fonts/OTF"),
|
||||||
|
Path("/usr/share/fonts/truetype"),
|
||||||
|
Path("/usr/share/fonts/opentype"),
|
||||||
|
Path("/usr/share/fonts/noto"),
|
||||||
|
Path("/usr/share/fonts/liberation"),
|
||||||
|
Path("/usr/share/fonts/gnu-free"),
|
||||||
|
Path("/usr/share/fonts/adobe-source-sans-pro"),
|
||||||
|
Path("/usr/share/fonts/adobe-source-sans"),
|
||||||
|
Path("/usr/share/fonts/inter"),
|
||||||
|
Path("/usr/share/fonts/cantarell"),
|
||||||
|
Path(os.path.expanduser("~/.local/share/fonts")),
|
||||||
|
Path(os.path.expanduser("~/.fonts")),
|
||||||
|
]
|
||||||
|
|
||||||
|
# (regular filename fragments, bold filename fragments) — first hit wins.
|
||||||
|
# Prefer clean sans families that CachyOS actually ships by default.
|
||||||
|
FONT_CANDIDATES = [
|
||||||
|
("DejaVuSans.ttf", "DejaVuSans-Bold.ttf"),
|
||||||
|
("DejaVuSans.otf", "DejaVuSans-Bold.otf"),
|
||||||
|
("NotoSans-Regular.ttf", "NotoSans-Bold.ttf"),
|
||||||
|
("NotoSans[wdth,wght].ttf", None), # variable; regular axis only
|
||||||
|
("LiberationSans-Regular.ttf", "LiberationSans-Bold.ttf"),
|
||||||
|
("LiberationSans-Regular.otf", "LiberationSans-Bold.otf"),
|
||||||
|
("SourceSans3-Regular.otf", "SourceSans3-Bold.otf"),
|
||||||
|
("SourceSans3-Regular.ttf", "SourceSans3-Bold.ttf"),
|
||||||
|
("SourceSansPro-Regular.otf", "SourceSansPro-Bold.otf"),
|
||||||
|
("Inter-Regular.ttf", "Inter-Bold.ttf"),
|
||||||
|
("Inter-Regular.otf", "Inter-Bold.otf"),
|
||||||
|
("FreeSans.otf", "FreeSansBold.otf"),
|
||||||
|
("FreeSans.ttf", "FreeSansBold.ttf"),
|
||||||
|
("Cantarell-Regular.otf", "Cantarell-Bold.otf"),
|
||||||
|
("Cantarell-VF.otf", None),
|
||||||
|
("NotoSans-Regular.otf", "NotoSans-Bold.otf"),
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def _index_font_files() -> dict[str, Path]:
|
||||||
|
"""Map lowercased basename → first matching path under FONT_ROOTS."""
|
||||||
|
found: dict[str, Path] = {}
|
||||||
|
seen_dirs: set[Path] = set()
|
||||||
|
for root in FONT_ROOTS:
|
||||||
|
if not root.is_dir():
|
||||||
|
continue
|
||||||
|
resolved = root.resolve()
|
||||||
|
if resolved in seen_dirs:
|
||||||
|
continue
|
||||||
|
seen_dirs.add(resolved)
|
||||||
|
for p in root.rglob("*"):
|
||||||
|
if p.suffix.lower() not in {".ttf", ".otf", ".ttc"}:
|
||||||
|
continue
|
||||||
|
if not p.is_file():
|
||||||
|
continue
|
||||||
|
key = p.name.lower()
|
||||||
|
found.setdefault(key, p)
|
||||||
|
return found
|
||||||
|
|
||||||
|
|
||||||
|
def register_fonts() -> str:
|
||||||
|
"""Register a UI sans for labels. Never crash if CachyOS lacks DejaVu."""
|
||||||
|
index = _index_font_files()
|
||||||
|
regular_path = None
|
||||||
|
bold_path = None
|
||||||
|
for regular_name, bold_name in FONT_CANDIDATES:
|
||||||
|
hit = index.get(regular_name.lower())
|
||||||
|
if hit is None:
|
||||||
|
continue
|
||||||
|
regular_path = hit
|
||||||
|
if bold_name:
|
||||||
|
bold_path = index.get(bold_name.lower())
|
||||||
|
break
|
||||||
|
|
||||||
|
if regular_path is None:
|
||||||
|
print("No system TTF/OTF sans found; using built-in Helvetica")
|
||||||
|
return "Helvetica"
|
||||||
|
|
||||||
|
pdfmetrics.registerFont(TTFont("SheetSans", str(regular_path)))
|
||||||
|
if bold_path is not None:
|
||||||
|
pdfmetrics.registerFont(TTFont("SheetSansBold", str(bold_path)))
|
||||||
|
print(f"Using font: {regular_path}")
|
||||||
|
if bold_path is not None:
|
||||||
|
print(f"Using bold: {bold_path}")
|
||||||
|
return "SheetSans"
|
||||||
|
|
||||||
|
|
||||||
|
def draw_crop_marks(c: canvas.Canvas, x: float, y: float, w: float, h: float,
|
||||||
|
length: float = 5 * mm, offset: float = 2 * mm) -> None:
|
||||||
|
"""Professional corner crop marks just outside the trim box."""
|
||||||
|
c.saveState()
|
||||||
|
c.setStrokeColorRGB(*MARK_COLOR)
|
||||||
|
c.setLineWidth(0.25)
|
||||||
|
# bottom-left
|
||||||
|
c.line(x - offset - length, y, x - offset, y)
|
||||||
|
c.line(x, y - offset - length, x, y - offset)
|
||||||
|
# bottom-right
|
||||||
|
c.line(x + w + offset, y, x + w + offset + length, y)
|
||||||
|
c.line(x + w, y - offset - length, x + w, y - offset)
|
||||||
|
# top-left
|
||||||
|
c.line(x - offset - length, y + h, x - offset, y + h)
|
||||||
|
c.line(x, y + h + offset, x, y + h + offset + length)
|
||||||
|
# top-right
|
||||||
|
c.line(x + w + offset, y + h, x + w + offset + length, y + h)
|
||||||
|
c.line(x + w, y + h + offset, x + w, y + h + offset + length)
|
||||||
|
c.restoreState()
|
||||||
|
|
||||||
|
|
||||||
|
def draw_rounded_guide(c: canvas.Canvas, x: float, y: float, w: float, h: float,
|
||||||
|
r: float) -> None:
|
||||||
|
"""Dashed rounded-rect sitting on the intended cut line."""
|
||||||
|
c.saveState()
|
||||||
|
c.setStrokeColorRGB(*GUIDE_COLOR)
|
||||||
|
c.setDash(1.2, 1.6)
|
||||||
|
c.setLineWidth(0.3)
|
||||||
|
c.roundRect(x, y, w, h, r, stroke=1, fill=0)
|
||||||
|
c.restoreState()
|
||||||
|
|
||||||
|
|
||||||
|
def layout_positions(cfg: dict) -> list[tuple[float, float, float, float]]:
|
||||||
|
"""Return list of (x, y, w, h) in points, origin bottom-left, centered on page."""
|
||||||
|
w = cfg["width_mm"] * mm
|
||||||
|
h = cfg["width_mm"] * ASPECT * mm
|
||||||
|
gap = cfg["gap_mm"] * mm
|
||||||
|
cols, rows = cfg["cols"], cfg["rows"]
|
||||||
|
|
||||||
|
block_w = cols * w + (cols - 1) * gap
|
||||||
|
block_h = rows * h + (rows - 1) * gap
|
||||||
|
|
||||||
|
# Center horizontally; keep a dedicated strip at the bottom for the legend.
|
||||||
|
margin = cfg["margin_mm"] * mm
|
||||||
|
legend_h = 16 * mm
|
||||||
|
origin_x = max(margin, (A4_W - block_w) / 2)
|
||||||
|
available_h = A4_H - legend_h
|
||||||
|
origin_y = legend_h + max(margin * 0.4, (available_h - block_h) / 2)
|
||||||
|
|
||||||
|
boxes = []
|
||||||
|
for r in range(rows):
|
||||||
|
for col in range(cols):
|
||||||
|
x = origin_x + col * (w + gap)
|
||||||
|
# rows from top visually → higher y is top
|
||||||
|
y = origin_y + (rows - 1 - r) * (h + gap)
|
||||||
|
boxes.append((x, y, w, h))
|
||||||
|
return boxes
|
||||||
|
|
||||||
|
|
||||||
|
def draw_legend(c: canvas.Canvas, cfg: dict, font: str, n_images: int) -> None:
|
||||||
|
c.saveState()
|
||||||
|
c.setFillColorRGB(*NOTE_COLOR)
|
||||||
|
c.setFont(font, 7)
|
||||||
|
y = 8 * mm
|
||||||
|
lines = [
|
||||||
|
f"A4 print sheet · {cfg['label']}",
|
||||||
|
f"Source {SRC_W}×{SRC_H} px placed 1:1 (no resampling) · "
|
||||||
|
f"print at 100% / Actual Size · do not 'fit to page'",
|
||||||
|
f"Grey marks = trim corners · dashed red = optional rounded cut "
|
||||||
|
f"(r = {cfg['corner_r_mm']:.1f} mm) · paper: 200–300 gsm card or photo stock",
|
||||||
|
]
|
||||||
|
for line in lines:
|
||||||
|
c.drawString(12 * mm, y, line)
|
||||||
|
y += 3.4 * mm
|
||||||
|
c.restoreState()
|
||||||
|
|
||||||
|
|
||||||
|
def draw_placeholder(c: canvas.Canvas, x, y, w, h, font: str, idx: int) -> None:
|
||||||
|
c.saveState()
|
||||||
|
c.setFillColorRGB(0.93, 0.93, 0.94)
|
||||||
|
c.rect(x, y, w, h, stroke=0, fill=1)
|
||||||
|
c.setStrokeColorRGB(0.7, 0.7, 0.72)
|
||||||
|
c.setLineWidth(0.4)
|
||||||
|
c.rect(x, y, w, h, stroke=1, fill=0)
|
||||||
|
# cross
|
||||||
|
c.setStrokeColorRGB(0.8, 0.8, 0.82)
|
||||||
|
c.line(x, y, x + w, y + h)
|
||||||
|
c.line(x, y + h, x + w, y)
|
||||||
|
c.setFillColorRGB(0.35, 0.35, 0.38)
|
||||||
|
c.setFont(font, 9)
|
||||||
|
c.drawCentredString(x + w / 2, y + h / 2 + 6, f"slot {idx + 1}")
|
||||||
|
c.setFont(font, 7)
|
||||||
|
c.drawCentredString(x + w / 2, y + h / 2 - 8, f"{SRC_W} × {SRC_H} px")
|
||||||
|
c.restoreState()
|
||||||
|
|
||||||
|
|
||||||
|
def compose_page(
|
||||||
|
c: canvas.Canvas,
|
||||||
|
cfg: dict,
|
||||||
|
images: list[Path | None],
|
||||||
|
font: str,
|
||||||
|
) -> None:
|
||||||
|
boxes = layout_positions(cfg)
|
||||||
|
# pad / cycle images to fill slots
|
||||||
|
slots = []
|
||||||
|
if images:
|
||||||
|
for i in range(len(boxes)):
|
||||||
|
slots.append(images[i % len(images)])
|
||||||
|
else:
|
||||||
|
slots = [None] * len(boxes)
|
||||||
|
|
||||||
|
for i, ((x, y, w, h), img_path) in enumerate(zip(boxes, slots)):
|
||||||
|
if img_path and img_path.exists():
|
||||||
|
# Draw the raster at the exact box size. ReportLab will embed the
|
||||||
|
# original pixels; because box mm == pixels / chosen-dpi, there is
|
||||||
|
# no interpolation of the source.
|
||||||
|
c.drawImage(
|
||||||
|
str(img_path),
|
||||||
|
x, y, width=w, height=h,
|
||||||
|
preserveAspectRatio=True,
|
||||||
|
anchor="c",
|
||||||
|
mask="auto",
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
draw_placeholder(c, x, y, w, h, font, i)
|
||||||
|
|
||||||
|
draw_crop_marks(c, x, y, w, h)
|
||||||
|
draw_rounded_guide(c, x, y, w, h, cfg["corner_r_mm"] * mm)
|
||||||
|
|
||||||
|
draw_legend(c, cfg, font, len([p for p in images if p]))
|
||||||
|
|
||||||
|
|
||||||
|
def build_pdf(out_path: Path, layout: str, image_paths: list[Path]) -> None:
|
||||||
|
font = register_fonts()
|
||||||
|
cfg = LAYOUTS[layout]
|
||||||
|
c = canvas.Canvas(str(out_path), pagesize=A4)
|
||||||
|
c.setTitle(f"Bookmark sheet ({layout})")
|
||||||
|
c.setAuthor("bookmark kit")
|
||||||
|
|
||||||
|
if not image_paths:
|
||||||
|
compose_page(c, cfg, [], font)
|
||||||
|
c.showPage()
|
||||||
|
else:
|
||||||
|
per_page = cfg["cols"] * cfg["rows"]
|
||||||
|
for start in range(0, len(image_paths), per_page):
|
||||||
|
chunk = image_paths[start : start + per_page]
|
||||||
|
compose_page(c, cfg, chunk, font)
|
||||||
|
c.showPage()
|
||||||
|
|
||||||
|
c.save()
|
||||||
|
print(f"Wrote {out_path}")
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||||
|
p.add_argument("images", nargs="*", type=Path, help="PNG/JPG sources (1008×1792 ideal)")
|
||||||
|
p.add_argument("--layout", choices=LAYOUTS, default="2up")
|
||||||
|
p.add_argument("--both", action="store_true", help="also write the other layout")
|
||||||
|
p.add_argument("-o", "--out", type=Path, default=None)
|
||||||
|
args = p.parse_args()
|
||||||
|
|
||||||
|
images = [ip for ip in args.images if ip.exists()]
|
||||||
|
missing = [ip for ip in args.images if not ip.exists()]
|
||||||
|
if missing:
|
||||||
|
print("Skipping missing files:", ", ".join(str(m) for m in missing))
|
||||||
|
|
||||||
|
layouts = [args.layout]
|
||||||
|
if args.both:
|
||||||
|
layouts = list(LAYOUTS)
|
||||||
|
|
||||||
|
for lay in layouts:
|
||||||
|
suffix = "template" if not images else "sheet"
|
||||||
|
out = args.out
|
||||||
|
if out is None or len(layouts) > 1:
|
||||||
|
out = Path(__file__).resolve().parent / f"bookmark_{lay}_{suffix}.pdf"
|
||||||
|
build_pdf(out, lay, images)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
@@ -0,0 +1,189 @@
|
|||||||
|
import argparse
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
|
||||||
|
"""
|
||||||
|
Very primitive and unpolished borg wrapper for certain actions.
|
||||||
|
|
||||||
|
It relies on a JSON config from which it reads borg vault locations and passwords.
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
def get_destination_host(destination_hosts, backup_destination):
|
||||||
|
for entry in destination_hosts:
|
||||||
|
if entry['alias'] == backup_destination:
|
||||||
|
return entry['host']
|
||||||
|
raise Exception(f"Destination host '{backup_destination}' is missing its configuration.")
|
||||||
|
|
||||||
|
|
||||||
|
def sync_backup(args):
|
||||||
|
operation = args.operation
|
||||||
|
|
||||||
|
print("Opening password file ...")
|
||||||
|
# Read Borg archive passwords from JSON file using sudo
|
||||||
|
password_file_path = '/root/borg_passwords.json'
|
||||||
|
password_command = f'sudo cat {password_file_path}'
|
||||||
|
password_data = os.popen(password_command).read().strip()
|
||||||
|
|
||||||
|
# Parse JSON data
|
||||||
|
passwords = json.loads(password_data)
|
||||||
|
|
||||||
|
if operation == 'create_all':
|
||||||
|
for entry in passwords['archives']:
|
||||||
|
password = entry['password']
|
||||||
|
destination_host_alias = entry['destination_host_alias']
|
||||||
|
destination_host = get_destination_host(passwords['destination_hosts'], destination_host_alias)
|
||||||
|
local_archive_root_path = entry['local_archive_root_path']
|
||||||
|
remote_archive_path = entry['remote_archive_path']
|
||||||
|
local_backup_source_dir = entry['local_backup_source_dir']
|
||||||
|
|
||||||
|
if args.backup_destination is not None and args.backup_destination != destination_host_alias:
|
||||||
|
print(f'skipping archive {entry['archive_name']} of destination {destination_host_alias}')
|
||||||
|
continue
|
||||||
|
if not args.skip_missing_source and not os.path.exists(local_archive_root_path):
|
||||||
|
raise RuntimeError(f"local archive root path '{local_archive_root_path}' doesn't exist")
|
||||||
|
elif args.skip_missing_source:
|
||||||
|
print(f"local archive root path is missing, skipping")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Set up Borg command with operation as the first argument
|
||||||
|
borg_env_var = f'export BORG_PASSPHRASE="{password}"'
|
||||||
|
borg_command = f'{borg_env_var}; cd {local_archive_root_path} && borg create --rsh \'ssh -i ~/.ssh/id_ed25519\''
|
||||||
|
|
||||||
|
if args.dry_run is True:
|
||||||
|
print("dry run enabled")
|
||||||
|
dry_run_or_stats = '--dry-run' if args.dry_run is True else '--stats'
|
||||||
|
if args.ignore_inode:
|
||||||
|
borg_command += ' --files-cache ctime,size' # ignore inode (useful for external disks)
|
||||||
|
borg_command += f' --iec --progress {dry_run_or_stats} -v --compression none --checkpoint-interval=900'
|
||||||
|
borg_command += f' {destination_host}{remote_archive_path}::' + '{utcnow}Z ' + f'{local_backup_source_dir}'
|
||||||
|
|
||||||
|
# Execute Borg command
|
||||||
|
print(f'[CMD] {";".join(borg_command.split(";")[1:])}')
|
||||||
|
retcode = subprocess.call(borg_command, shell=True)
|
||||||
|
if retcode != 0:
|
||||||
|
raise RuntimeError("exited with non-zero return code")
|
||||||
|
elif operation == 'prune_all':
|
||||||
|
for entry in passwords['archives']:
|
||||||
|
password = entry['password']
|
||||||
|
destination_host = get_destination_host(passwords['destination_hosts'], entry['destination_host_alias'])
|
||||||
|
local_archive_root_path = entry['local_archive_root_path']
|
||||||
|
remote_archive_path = entry['remote_archive_path']
|
||||||
|
#local_backup_source_dir = entry['local_backup_source_dir']
|
||||||
|
|
||||||
|
if not args.skip_missing_source and not os.path.exists(local_archive_root_path):
|
||||||
|
raise RuntimeError(f"local archive root path '{local_archive_root_path}' doesn't exist")
|
||||||
|
elif args.skip_missing_source:
|
||||||
|
print(f"local archive root path is missing, skipping")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Set up Borg command with operation as the first argument
|
||||||
|
borg_env_var = f'export BORG_PASSPHRASE="{password}"'
|
||||||
|
borg_command = f'{borg_env_var}; cd {local_archive_root_path} && borg prune --rsh \'ssh -i ~/.ssh/id_ed25519\''
|
||||||
|
|
||||||
|
if args.dry_run is True:
|
||||||
|
print("dry run enabled")
|
||||||
|
dry_run_or_stats = '--dry-run' if args.dry_run is True else '--stats'
|
||||||
|
borg_command += f' --iec --progress {dry_run_or_stats} -v'
|
||||||
|
borg_command += f' --list --keep-last 1 {destination_host}{remote_archive_path}'
|
||||||
|
|
||||||
|
# Execute Borg command
|
||||||
|
print(f'[CMD] {";".join(borg_command.split(";")[1:])}')
|
||||||
|
retcode = subprocess.call(borg_command, shell=True)
|
||||||
|
if retcode != 0:
|
||||||
|
raise RuntimeError("exited with non-zero return code")
|
||||||
|
elif operation == 'compact_all':
|
||||||
|
for entry in passwords['archives']:
|
||||||
|
password = entry['password']
|
||||||
|
destination_host = get_destination_host(passwords['destination_hosts'], entry['destination_host_alias'])
|
||||||
|
local_archive_root_path = entry['local_archive_root_path']
|
||||||
|
remote_archive_path = entry['remote_archive_path']
|
||||||
|
#local_backup_source_dir = entry['local_backup_source_dir']
|
||||||
|
|
||||||
|
if not args.skip_missing_source and not os.path.exists(local_archive_root_path):
|
||||||
|
raise RuntimeError(f"local archive root path '{local_archive_root_path}' doesn't exist")
|
||||||
|
elif args.skip_missing_source:
|
||||||
|
print(f"local archive root path is missing, skipping")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Set up Borg command with operation as the first argument
|
||||||
|
borg_env_var = f'export BORG_PASSPHRASE="{password}"'
|
||||||
|
borg_command = f'{borg_env_var}; cd {local_archive_root_path} && borg compact --rsh \'ssh -i ~/.ssh/id_ed25519\''
|
||||||
|
|
||||||
|
borg_command += f' --iec --progress -v {destination_host}{remote_archive_path}'
|
||||||
|
|
||||||
|
# Execute Borg command
|
||||||
|
print(f'[CMD] {";".join(borg_command.split(";")[1:])}')
|
||||||
|
retcode = subprocess.call(borg_command, shell=True)
|
||||||
|
if retcode != 0:
|
||||||
|
raise RuntimeError("exited with non-zero return code")
|
||||||
|
else:
|
||||||
|
found = False
|
||||||
|
for entry in passwords['archives']:
|
||||||
|
if entry['archive_name'] == args.archive_name and entry['destination_host_alias'] == args.backup_destination:
|
||||||
|
found = True
|
||||||
|
password = entry['password']
|
||||||
|
destination_host = get_destination_host(passwords['destination_hosts'], args.backup_destination)
|
||||||
|
local_archive_root_path = entry['local_archive_root_path']
|
||||||
|
remote_archive_path = entry['remote_archive_path']
|
||||||
|
local_backup_source_dir = entry['local_backup_source_dir']
|
||||||
|
|
||||||
|
if not args.skip_missing_source and not os.path.exists(local_archive_root_path):
|
||||||
|
raise RuntimeError(f"local archive root path '{local_archive_root_path}' doesn't exist")
|
||||||
|
elif args.skip_missing_source:
|
||||||
|
print(f"local archive root path is missing, skipping")
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Set up Borg command with operation as the first argument
|
||||||
|
borg_env_var = f'export BORG_PASSPHRASE="{password}"'
|
||||||
|
borg_command = f'{borg_env_var}; cd {local_archive_root_path} && borg {operation} --rsh \'ssh -i ~/.ssh/id_ed25519\''
|
||||||
|
|
||||||
|
if operation == 'create':
|
||||||
|
dry_run_or_stats = '--dry-run' if args.dry_run is True else '--stats'
|
||||||
|
if args.ignore_inode:
|
||||||
|
borg_command += ' --files-cache ctime,size' # ignore inode (useful for external disks)
|
||||||
|
if not args.skip_list:
|
||||||
|
borg_command += ' --list'
|
||||||
|
borg_command += f' --iec --progress {dry_run_or_stats} -v --compression none --checkpoint-interval=900'
|
||||||
|
borg_command += f' {destination_host}{remote_archive_path}::' + '{utcnow}Z ' + f'{local_backup_source_dir}'
|
||||||
|
elif operation == 'list':
|
||||||
|
borg_command += f' {destination_host}{remote_archive_path}'
|
||||||
|
elif operation == 'info':
|
||||||
|
borg_command += f' --iec {destination_host}{remote_archive_path}'
|
||||||
|
elif operation == 'prune':
|
||||||
|
borg_command += f' --list --keep-last 1 {destination_host}{remote_archive_path}'
|
||||||
|
elif operation == 'compact':
|
||||||
|
borg_command += f' --iec --progress {destination_host}{remote_archive_path}'
|
||||||
|
# elif operation == 'delete':
|
||||||
|
else:
|
||||||
|
print(f"Invalid operation '{operation}'")
|
||||||
|
return
|
||||||
|
|
||||||
|
# Execute Borg command
|
||||||
|
print(f'[CMD] {";".join(borg_command.split(";")[1:])}')
|
||||||
|
os.system(borg_command)
|
||||||
|
break
|
||||||
|
if not found:
|
||||||
|
print(f"Couldn't find a pair of archive '{args.archive_name}' and destination '{args.backup_destination}'.")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
parser = argparse.ArgumentParser(description="Sync backups using Borg via CLI")
|
||||||
|
parser.add_argument("operation", type=str, help="Borg operation (create, list, prune, info, compact, create_all, prune_all, compact_all)")
|
||||||
|
parser.add_argument("-a", "--archive-name", type=str, help="Name of the Borg archive", required=False)
|
||||||
|
parser.add_argument("-d", "--backup-destination", type=str, help="Destination for the backup", required=False)
|
||||||
|
parser.add_argument("--dry-run", type=bool, help="If specified, shows only what would happen", required=False)
|
||||||
|
parser.add_argument("--skip-missing-source", action=argparse.BooleanOptionalAction, help="Skips syncing a missing archive instead of crashing", required=False)
|
||||||
|
parser.add_argument("--ignore-inode", action=argparse.BooleanOptionalAction, help="Ignores inodes when creating a backup", required=False)
|
||||||
|
parser.add_argument("--skip-list", action=argparse.BooleanOptionalAction, help="Skips printing list of synced files", required=False)
|
||||||
|
arguments = parser.parse_args()
|
||||||
|
print(f'debug: {arguments.operation}')
|
||||||
|
if arguments.operation in ["create_all", "prune_all", "compact_all"]:
|
||||||
|
if arguments.archive_name is not None:
|
||||||
|
raise RuntimeError(f"Argument -a cannot be used with operation '{arguments.operation}'.")
|
||||||
|
else:
|
||||||
|
if arguments.archive_name is None or arguments.backup_destination is None:
|
||||||
|
raise RuntimeError(f"Arguments -a and -d must be used with operation '{arguments.operation}'.")
|
||||||
|
sync_backup(arguments)
|
||||||
|
|
||||||
@@ -0,0 +1,19 @@
|
|||||||
|
{
|
||||||
|
"destination_hosts": [
|
||||||
|
{
|
||||||
|
"alias": "your-host",
|
||||||
|
"host": "ssh://your-server.example.com:22"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"archives": [
|
||||||
|
{
|
||||||
|
"archive_name": "arc01",
|
||||||
|
"destination_host_alias": "your-host",
|
||||||
|
"remote_archive_path": "/home/remote-user/archive01",
|
||||||
|
"password": "placeholder",
|
||||||
|
"local_archive_root_path": "/home/local-user/archive01/",
|
||||||
|
"local_backup_source_dir": "./"
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
Executable
+195
@@ -0,0 +1,195 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
import argparse
|
||||||
|
import hashlib
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from collections import defaultdict
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
CHUNK = 1024
|
||||||
|
|
||||||
|
G = "\033[32m"
|
||||||
|
Y = "\033[33m"
|
||||||
|
C = "\033[36m"
|
||||||
|
R = "\033[31m"
|
||||||
|
B = "\033[1m"
|
||||||
|
Z = "\033[0m"
|
||||||
|
|
||||||
|
|
||||||
|
def parse_size(s: str) -> int:
|
||||||
|
s = s.strip().lower()
|
||||||
|
mul = 1
|
||||||
|
for suf, m in (
|
||||||
|
("kb", 1024), ("k", 1024),
|
||||||
|
("mb", 1024**2), ("m", 1024**2),
|
||||||
|
("gb", 1024**3), ("g", 1024**3),
|
||||||
|
("b", 1),
|
||||||
|
):
|
||||||
|
if s.endswith(suf):
|
||||||
|
s = s[:-len(suf)].strip()
|
||||||
|
mul = m
|
||||||
|
break
|
||||||
|
return int(float(s) * mul)
|
||||||
|
|
||||||
|
|
||||||
|
def human(n: int) -> str:
|
||||||
|
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
|
||||||
|
if n < 1024 or unit == "TiB":
|
||||||
|
return f"{n:.0f}{unit}" if unit == "B" else f"{n:.1f}{unit}"
|
||||||
|
n /= 1024
|
||||||
|
|
||||||
|
|
||||||
|
def prog(stage: str, cur: int, total: int) -> None:
|
||||||
|
msg = f"{stage:<9} | {cur}/{total} files"
|
||||||
|
if not hasattr(prog, "_last"):
|
||||||
|
prog._last = 0
|
||||||
|
pad = max(0, prog._last - len(msg))
|
||||||
|
sys.stderr.write("\r" + C + msg + Z + (" " * pad))
|
||||||
|
sys.stderr.flush()
|
||||||
|
prog._last = len(msg)
|
||||||
|
|
||||||
|
|
||||||
|
def parse_args():
|
||||||
|
ap = argparse.ArgumentParser()
|
||||||
|
ap.add_argument("root", nargs="?", default=".")
|
||||||
|
ap.add_argument("--min-size", default="1")
|
||||||
|
ap.add_argument("--max-size", default=None)
|
||||||
|
ap.add_argument("--ignore", action="append", default=[])
|
||||||
|
return ap.parse_args()
|
||||||
|
|
||||||
|
|
||||||
|
def is_ignored(p: Path, ignore: list[Path]) -> bool:
|
||||||
|
try:
|
||||||
|
rp = p.resolve()
|
||||||
|
except OSError:
|
||||||
|
return False
|
||||||
|
return any(rp == q or q in rp.parents for q in ignore)
|
||||||
|
|
||||||
|
|
||||||
|
def file_head_tail(p: Path, n: int = CHUNK):
|
||||||
|
try:
|
||||||
|
size = p.stat().st_size
|
||||||
|
with p.open("rb") as f:
|
||||||
|
head = f.read(n)
|
||||||
|
f.seek(max(0, size - n))
|
||||||
|
tail = f.read(n)
|
||||||
|
return head, tail
|
||||||
|
except OSError:
|
||||||
|
return None, None
|
||||||
|
|
||||||
|
|
||||||
|
def sha256sum(p: Path) -> str | None:
|
||||||
|
h = hashlib.sha256()
|
||||||
|
try:
|
||||||
|
with p.open("rb") as f:
|
||||||
|
while True:
|
||||||
|
b = f.read(1024 * 1024)
|
||||||
|
if not b:
|
||||||
|
break
|
||||||
|
h.update(b)
|
||||||
|
return h.hexdigest()
|
||||||
|
except OSError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def iter_files(root: Path, ignore: list[Path], min_size: int, max_size: int | None):
|
||||||
|
for dirpath, dirnames, filenames in os.walk(root):
|
||||||
|
dp = Path(dirpath)
|
||||||
|
dirnames[:] = [d for d in dirnames if not (dp / d).is_symlink() and not is_ignored(dp / d, ignore)]
|
||||||
|
|
||||||
|
for name in filenames:
|
||||||
|
p = dp / name
|
||||||
|
if p.is_symlink() or is_ignored(p, ignore):
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
size = p.stat().st_size
|
||||||
|
except OSError:
|
||||||
|
continue
|
||||||
|
if size < min_size:
|
||||||
|
continue
|
||||||
|
if max_size is not None and size > max_size:
|
||||||
|
continue
|
||||||
|
yield p, size
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
args = parse_args()
|
||||||
|
root = Path(args.root).expanduser().resolve()
|
||||||
|
min_size = parse_size(args.min_size)
|
||||||
|
max_size = parse_size(args.max_size) if args.max_size else None
|
||||||
|
ignore = [Path(p).expanduser().resolve() for p in args.ignore]
|
||||||
|
|
||||||
|
by_size = defaultdict(list)
|
||||||
|
total = 0
|
||||||
|
|
||||||
|
for p, size in iter_files(root, ignore, min_size, max_size):
|
||||||
|
total += 1
|
||||||
|
if total % 100 == 0:
|
||||||
|
prog("Scanning", total, total)
|
||||||
|
by_size[size].append(p)
|
||||||
|
|
||||||
|
if not total:
|
||||||
|
print("No files found.")
|
||||||
|
return
|
||||||
|
|
||||||
|
prog("Scanning", total, total)
|
||||||
|
sys.stderr.write("\n")
|
||||||
|
|
||||||
|
candidates = {s: ps for s, ps in by_size.items() if len(ps) > 1}
|
||||||
|
if not candidates:
|
||||||
|
print("No duplicates found.")
|
||||||
|
return
|
||||||
|
|
||||||
|
by_ht = defaultdict(list)
|
||||||
|
ht_total = sum(len(ps) for ps in candidates.values())
|
||||||
|
done = 0
|
||||||
|
for size, ps in candidates.items():
|
||||||
|
for p in ps:
|
||||||
|
done += 1
|
||||||
|
prog("Head/tail", done, ht_total)
|
||||||
|
h, t = file_head_tail(p)
|
||||||
|
if h is not None:
|
||||||
|
by_ht[(size, h, t)].append(p)
|
||||||
|
sys.stderr.write("\n")
|
||||||
|
|
||||||
|
ht_groups = {k: ps for k, ps in by_ht.items() if len(ps) > 1}
|
||||||
|
if not ht_groups:
|
||||||
|
print("No duplicates found.")
|
||||||
|
return
|
||||||
|
|
||||||
|
by_hash = defaultdict(list)
|
||||||
|
hash_total = sum(len(ps) for ps in ht_groups.values())
|
||||||
|
done = 0
|
||||||
|
for ps in ht_groups.values():
|
||||||
|
for p in ps:
|
||||||
|
done += 1
|
||||||
|
prog("Hashing", done, hash_total)
|
||||||
|
h = sha256sum(p)
|
||||||
|
if h:
|
||||||
|
by_hash[h].append(p)
|
||||||
|
sys.stderr.write("\n")
|
||||||
|
|
||||||
|
groups = []
|
||||||
|
for h, ps in by_hash.items():
|
||||||
|
if len(ps) > 1:
|
||||||
|
try:
|
||||||
|
size = ps[0].stat().st_size
|
||||||
|
except OSError:
|
||||||
|
size = 0
|
||||||
|
groups.append((len(ps), size, h, ps))
|
||||||
|
|
||||||
|
if not groups:
|
||||||
|
print("No duplicates found.")
|
||||||
|
return
|
||||||
|
|
||||||
|
groups.sort(key=lambda x: (-x[0], -x[1]))
|
||||||
|
|
||||||
|
for idx, (count, size, h, ps) in enumerate(groups, 1):
|
||||||
|
print(f"{B}{Y}Group {idx}{Z} ({count} files, {human(size)}) {C}{h[:12]}{Z}")
|
||||||
|
for p in sorted(ps):
|
||||||
|
print(f" {G}{p.resolve()}{Z}")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
||||||
@@ -10,7 +10,7 @@
|
|||||||
Apply a warm print color correction (less blue, slightly yellower).
|
Apply a warm print color correction (less blue, slightly yellower).
|
||||||
|
|
||||||
Usage with uv (no venv, no uv pip):
|
Usage with uv (no venv, no uv pip):
|
||||||
uv run warm_print.py /path/to/image.jpg
|
uv run warm_print.py image.jpg other.jpg /path/to/more.jpg
|
||||||
|
|
||||||
uv reads the dependencies in the header above and installs them
|
uv reads the dependencies in the header above and installs them
|
||||||
into an isolated cache on first run.
|
into an isolated cache on first run.
|
||||||
@@ -83,17 +83,8 @@ def warm_correct(im: Image.Image) -> Image.Image:
|
|||||||
return out_im
|
return out_im
|
||||||
|
|
||||||
|
|
||||||
def main() -> int:
|
def process_one(src: Path) -> Path:
|
||||||
if len(sys.argv) < 2:
|
"""Color-correct one image and write sibling *_warmprint file."""
|
||||||
print("Usage: python warm_print.py /path/to/image.jpg", file=sys.stderr)
|
|
||||||
return 2
|
|
||||||
|
|
||||||
src = Path(sys.argv[1]).expanduser().resolve()
|
|
||||||
if not src.is_file():
|
|
||||||
print(f"Not a file: {src}", file=sys.stderr)
|
|
||||||
return 1
|
|
||||||
|
|
||||||
# Same folder, same name, short suffix before the original extension.
|
|
||||||
dest = src.with_name(f"{src.stem}_warmprint{src.suffix}")
|
dest = src.with_name(f"{src.stem}_warmprint{src.suffix}")
|
||||||
|
|
||||||
im = Image.open(src)
|
im = Image.open(src)
|
||||||
@@ -112,8 +103,27 @@ def main() -> int:
|
|||||||
out_im.save(dest, "JPEG", **save_kwargs)
|
out_im.save(dest, "JPEG", **save_kwargs)
|
||||||
else:
|
else:
|
||||||
out_im.save(dest, "PNG", **save_kwargs)
|
out_im.save(dest, "PNG", **save_kwargs)
|
||||||
|
return dest
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
if len(sys.argv) < 2:
|
||||||
|
print(
|
||||||
|
"Usage: python warm_print.py image.jpg [more.jpg ...]",
|
||||||
|
file=sys.stderr,
|
||||||
|
)
|
||||||
|
return 2
|
||||||
|
|
||||||
|
failed = 0
|
||||||
|
for raw in sys.argv[1:]:
|
||||||
|
src = Path(raw).expanduser().resolve()
|
||||||
|
if not src.is_file():
|
||||||
|
print(f"Not a file: {src}", file=sys.stderr)
|
||||||
|
failed += 1
|
||||||
|
continue
|
||||||
|
dest = process_one(src)
|
||||||
print(dest)
|
print(dest)
|
||||||
return 0
|
return 1 if failed else 0
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
Reference in new issue
Block a user