Compare commits
2
Commits
main
..
17d8237022
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
17d8237022
|
||
|
|
bb1742416a |
No files matched your search
@@ -1 +0,0 @@
|
|||||||
*.pdf
|
|
||||||
@@ -1,351 +0,0 @@
|
|||||||
#!/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()
|
|
||||||
@@ -1,189 +0,0 @@
|
|||||||
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)
|
|
||||||
|
|
||||||
@@ -1,19 +0,0 @@
|
|||||||
{
|
|
||||||
"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": "./"
|
|
||||||
}
|
|
||||||
]
|
|
||||||
}
|
|
||||||
|
|
||||||
@@ -1,195 +0,0 @@
|
|||||||
#!/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 image.jpg other.jpg /path/to/more.jpg
|
uv run warm_print.py /path/to/image.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,8 +83,17 @@ def warm_correct(im: Image.Image) -> Image.Image:
|
|||||||
return out_im
|
return out_im
|
||||||
|
|
||||||
|
|
||||||
def process_one(src: Path) -> Path:
|
def main() -> int:
|
||||||
"""Color-correct one image and write sibling *_warmprint file."""
|
if len(sys.argv) < 2:
|
||||||
|
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)
|
||||||
@@ -103,27 +112,8 @@ def process_one(src: Path) -> Path:
|
|||||||
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
|
print(dest)
|
||||||
|
return 0
|
||||||
|
|
||||||
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)
|
|
||||||
return 1 if failed else 0
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
Reference in new issue
Block a user