refactor: rename package from mytoolkit to bin
- Move all source files from mytoolkit/ to bin/ - Update entry point and build config in pyproject.toml - Add pillow and pypdf dependencies - Update README and .gitignore paths - Remove unused subprocess import in pdf.py - Clean up duplicate imports in pdf merge command
This commit is contained in:
Zhengshou Lai
@@ -0,0 +1,9 @@
|
||||
"""MyCLI subcommands."""
|
||||
|
||||
from . import pdf, image, latex, video, bib, utils
|
||||
from .env import env_cmd
|
||||
from .ssh import ssh_cmd
|
||||
from .git import git_cmd
|
||||
from .self_mgmt import self_cmd, update_cmd, uninstall_cmd
|
||||
|
||||
__all__ = ["pdf", "image", "latex", "video", "bib", "utils", "env_cmd", "ssh_cmd", "git_cmd", "self_cmd", "update_cmd", "uninstall_cmd"]
|
||||
@@ -0,0 +1,56 @@
|
||||
"""BibTeX utilities."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from bin.config import config
|
||||
from bin.utils import handle_errors, run_command
|
||||
|
||||
|
||||
@click.group()
|
||||
def bib():
|
||||
"""BibTeX and bibliography commands."""
|
||||
pass
|
||||
|
||||
|
||||
@bib.command("to-markdown")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def to_markdown(files, dry_run):
|
||||
"""Convert BibTeX files to Markdown."""
|
||||
for pattern in files:
|
||||
for bib_file in Path(".").glob(pattern):
|
||||
if bib_file.suffix.lower() != ".bib":
|
||||
continue
|
||||
output = bib_file.with_suffix(".md")
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would convert: {bib_file} -> {output}")
|
||||
continue
|
||||
|
||||
click.echo(f"Converting: {bib_file} -> {output}")
|
||||
content = bib_file.read_text()
|
||||
md_content = f"# Bibliography\n\n```bibtex\n{content}\n```\n"
|
||||
output.write_text(md_content)
|
||||
|
||||
|
||||
@bib.command("cv-update")
|
||||
@handle_errors
|
||||
def cv_update():
|
||||
"""Update CV bibliography using biber."""
|
||||
cv_path_str = config.get("path_cv")
|
||||
if cv_path_str:
|
||||
cv_path = Path(cv_path_str).expanduser()
|
||||
else:
|
||||
# Fallback to hardcoded path
|
||||
cv_path = Path.home() / "Documents/Personal/resume_cv/cv_lai_maintaining_latex/cv_lai_maintaining"
|
||||
|
||||
if not cv_path.exists():
|
||||
click.echo(f"CV file not found: {cv_path}")
|
||||
return
|
||||
|
||||
click.echo(f"Updating CV bibliography: {cv_path}")
|
||||
run_command(["biber", str(cv_path)], check=False)
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Environment variables management."""
|
||||
|
||||
import click
|
||||
|
||||
from bin.config import config
|
||||
|
||||
|
||||
@click.group(name="env")
|
||||
def env_cmd():
|
||||
"""Manage environment variables."""
|
||||
pass
|
||||
|
||||
|
||||
@env_cmd.command("list")
|
||||
def env_list():
|
||||
"""List all vars."""
|
||||
vars = config.get_all()
|
||||
if not vars:
|
||||
click.echo("No vars configured.")
|
||||
return
|
||||
for name, value in sorted(vars.items()):
|
||||
masked = "***" if any(x in name.lower() for x in ["key", "secret", "token", "pass"]) else value
|
||||
click.echo(f"{name:20} = {masked}")
|
||||
|
||||
|
||||
@env_cmd.command("get")
|
||||
@click.argument("name")
|
||||
def env_get(name):
|
||||
"""Get a var value."""
|
||||
value = config.get(name)
|
||||
if value is None:
|
||||
click.echo(f"Var not found: {name}", err=True)
|
||||
raise click.Exit(1)
|
||||
click.echo(value)
|
||||
|
||||
|
||||
@env_cmd.command("set")
|
||||
@click.argument("name")
|
||||
@click.argument("value", required=False)
|
||||
@click.option("--secret", is_flag=True, help="Hide input for secrets")
|
||||
def env_set(name, value, secret):
|
||||
"""Set a var. Prompts if value not provided."""
|
||||
if value is None:
|
||||
value = click.prompt("Value", hide_input=secret)
|
||||
config.set(name, value)
|
||||
click.secho(f"Set: {name}", fg="green")
|
||||
|
||||
|
||||
@env_cmd.command("remove")
|
||||
@click.argument("name")
|
||||
@click.confirmation_option(prompt="Remove this var?")
|
||||
def env_remove(name):
|
||||
"""Remove a var."""
|
||||
if config.remove(name):
|
||||
click.secho(f"Removed: {name}", fg="green")
|
||||
else:
|
||||
click.echo(f"Var not found: {name}", err=True)
|
||||
raise click.Exit(1)
|
||||
|
||||
|
||||
@env_cmd.command("export")
|
||||
def env_export():
|
||||
"""Print export statements for shell eval."""
|
||||
for name, value in config.export().items():
|
||||
click.echo(f"export {name}={value}")
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Git utilities including proxy management."""
|
||||
|
||||
import subprocess
|
||||
|
||||
import click
|
||||
|
||||
from bin.config import config
|
||||
|
||||
|
||||
def _get_proxies() -> dict[str, str]:
|
||||
"""Get git proxies from config."""
|
||||
proxies = {}
|
||||
for key, value in config.get_all().items():
|
||||
if key.startswith("proxy_"):
|
||||
name = key[6:] # Remove 'proxy_' prefix
|
||||
proxies[name] = value
|
||||
return proxies
|
||||
|
||||
|
||||
def _get_git_proxy(key: str) -> str:
|
||||
"""Get a global git proxy config value."""
|
||||
result = subprocess.run(
|
||||
["git", "config", "--global", key],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
return result.stdout.strip() if result.returncode == 0 else "(not set)"
|
||||
|
||||
|
||||
def _unset_git_proxy(key: str) -> None:
|
||||
"""Unset a global git proxy config value (no-op if not set)."""
|
||||
subprocess.run(
|
||||
["git", "config", "--global", "--unset", key],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
@click.group(name="git")
|
||||
def git_cmd() -> None:
|
||||
"""Git utilities."""
|
||||
pass
|
||||
|
||||
|
||||
@git_cmd.group(name="proxy")
|
||||
def proxy_cmd() -> None:
|
||||
"""Manage git proxy settings."""
|
||||
pass
|
||||
|
||||
|
||||
def _complete_proxy_names(_ctx, _param, incomplete):
|
||||
"""Shell completion for proxy names."""
|
||||
return [
|
||||
name for name in _get_proxies() if name.startswith(incomplete)
|
||||
]
|
||||
|
||||
|
||||
@proxy_cmd.command("set")
|
||||
@click.argument("name", shell_complete=_complete_proxy_names)
|
||||
def proxy_set(name: str) -> None:
|
||||
"""Set git proxy by name."""
|
||||
proxies = _get_proxies()
|
||||
if name not in proxies:
|
||||
click.echo(f"Unknown proxy: {name}", err=True)
|
||||
click.echo(f"Available: {', '.join(proxies.keys())}", err=True)
|
||||
raise click.Exit(1)
|
||||
|
||||
proxy_url = proxies[name]
|
||||
subprocess.run(
|
||||
["git", "config", "--global", "http.proxy", proxy_url], check=True
|
||||
)
|
||||
subprocess.run(
|
||||
["git", "config", "--global", "https.proxy", proxy_url], check=True
|
||||
)
|
||||
click.secho(f"Git proxy set to {name}: {proxy_url}", fg="green")
|
||||
|
||||
|
||||
@proxy_cmd.command("unset")
|
||||
def proxy_unset() -> None:
|
||||
"""Unset git proxy."""
|
||||
_unset_git_proxy("http.proxy")
|
||||
_unset_git_proxy("https.proxy")
|
||||
click.secho("Git proxy unset", fg="green")
|
||||
|
||||
|
||||
@proxy_cmd.command("status")
|
||||
def proxy_status() -> None:
|
||||
"""Show current git proxy status."""
|
||||
http_proxy = _get_git_proxy("http.proxy")
|
||||
https_proxy = _get_git_proxy("https.proxy")
|
||||
|
||||
proxies = _get_proxies()
|
||||
current_name = next(
|
||||
(p for p, url in proxies.items() if url == http_proxy), None
|
||||
)
|
||||
|
||||
click.secho("Git proxy status:", fg="cyan")
|
||||
if current_name:
|
||||
click.echo(f" Active: {current_name}")
|
||||
click.echo(f" http.proxy: {http_proxy}")
|
||||
click.echo(f" https.proxy: {https_proxy}")
|
||||
|
||||
click.secho("\nAvailable proxies:", fg="cyan")
|
||||
for proxy_name, url in proxies.items():
|
||||
marker = "*" if proxy_name == current_name else " "
|
||||
click.echo(f" [{marker}] {proxy_name:12} {url}")
|
||||
|
||||
|
||||
@proxy_cmd.command("list")
|
||||
def proxy_list_cmd() -> None:
|
||||
"""List available git proxies."""
|
||||
proxies = _get_proxies()
|
||||
click.secho("Available proxies:", fg="cyan")
|
||||
for name, url in proxies.items():
|
||||
click.echo(f" {name:12} {url}")
|
||||
@@ -0,0 +1,183 @@
|
||||
"""Image utilities."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from bin.utils import handle_errors, run_command
|
||||
|
||||
|
||||
@click.group()
|
||||
def image():
|
||||
"""Image manipulation commands."""
|
||||
pass
|
||||
|
||||
|
||||
@image.command("eps-to-pdf")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def eps_to_pdf(files, dry_run):
|
||||
"""Convert EPS files to PDF."""
|
||||
for pattern in files:
|
||||
for eps_file in Path(".").glob(pattern):
|
||||
if eps_file.suffix.lower() not in (".eps", ".ps"):
|
||||
continue
|
||||
output = eps_file.with_suffix(".pdf")
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would convert: {eps_file} -> {output}")
|
||||
continue
|
||||
|
||||
click.echo(f"Converting: {eps_file} -> {output}")
|
||||
run_command(["epstopdf", str(eps_file)], check=False)
|
||||
|
||||
|
||||
@image.command("eps-fix")
|
||||
@click.argument("file", type=click.Path(exists=True))
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def eps_fix(file, dry_run):
|
||||
"""Fix EPS files created by print command."""
|
||||
path = Path(file)
|
||||
content = path.read_text()
|
||||
|
||||
if "/f/fill" not in content:
|
||||
click.echo("File already fixed or not created by print command")
|
||||
return
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would fix: {file}")
|
||||
return
|
||||
|
||||
lines = content.split("\n")
|
||||
new_lines = []
|
||||
for line in lines:
|
||||
if "/f/fill" in line:
|
||||
line = line.replace("/f/fill", "")
|
||||
new_lines.append(line)
|
||||
|
||||
path.write_text("\n".join(new_lines))
|
||||
click.echo(f"Fixed: {file}")
|
||||
|
||||
|
||||
@image.command("jpg-to-pdf")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--output", "-o", help="Output PDF file")
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def jpg_to_pdf(files, output, dry_run):
|
||||
"""Convert JPG files to PDF."""
|
||||
images = []
|
||||
for pattern in files:
|
||||
for jpg_file in Path(".").glob(pattern):
|
||||
if jpg_file.suffix.lower() in (".jpg", ".jpeg"):
|
||||
images.append(str(jpg_file))
|
||||
|
||||
if not images:
|
||||
click.echo("No JPG files found")
|
||||
return
|
||||
|
||||
output = output or "output.pdf"
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would convert {len(images)} images to {output}")
|
||||
return
|
||||
|
||||
click.echo(f"Converting {len(images)} images to {output}")
|
||||
run_command(["convert"] + images + [output], check=False)
|
||||
|
||||
|
||||
@image.command("tiff-to-pdf")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def tiff_to_pdf(files, dry_run):
|
||||
"""Convert TIFF files to PDF."""
|
||||
for pattern in files:
|
||||
for tiff_file in Path(".").glob(pattern):
|
||||
if tiff_file.suffix.lower() not in (".tif", ".tiff"):
|
||||
continue
|
||||
output = tiff_file.with_suffix(".pdf")
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would convert: {tiff_file} -> {output}")
|
||||
continue
|
||||
|
||||
click.echo(f"Converting: {tiff_file} -> {output}")
|
||||
run_command(["convert", str(tiff_file), str(output)], check=False)
|
||||
|
||||
|
||||
@image.command("tiff-compress")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def tiff_compress(files, dry_run):
|
||||
"""Compress TIFF files."""
|
||||
for pattern in files:
|
||||
for tiff_file in Path(".").glob(pattern):
|
||||
if tiff_file.suffix.lower() not in (".tif", ".tiff"):
|
||||
continue
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would compress: {tiff_file}")
|
||||
continue
|
||||
|
||||
click.echo(f"Compressing: {tiff_file}")
|
||||
run_command(
|
||||
["convert", str(tiff_file), "-compress", "zip", str(tiff_file)],
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
@image.command("compress")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--target-mb", "-t", default=1.0, help="Target max file size in MB")
|
||||
@click.option("--max-width", "-w", default=1600, help="Max width in pixels if resize needed")
|
||||
@click.option("--quality", "-q", default=75, help="JPEG quality for resized images")
|
||||
@click.option("--suffix", "-s", default="_compressed", help="Output filename suffix")
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def compress_images(files, target_mb, max_width, quality, suffix, dry_run):
|
||||
"""Compress PNG/JPG images to fit under target size."""
|
||||
from PIL import Image
|
||||
|
||||
for pattern in files:
|
||||
for img_file in Path(".").glob(pattern):
|
||||
if img_file.suffix.lower() not in (".png", ".jpg", ".jpeg"):
|
||||
continue
|
||||
|
||||
output = img_file.with_stem(f"{img_file.stem}{suffix}")
|
||||
if img_file.suffix.lower() == ".png":
|
||||
output = output.with_suffix(".jpg")
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would compress: {img_file} -> {output}")
|
||||
continue
|
||||
|
||||
img = Image.open(img_file)
|
||||
if img.mode in ("RGBA", "P"):
|
||||
img = img.convert("RGB")
|
||||
|
||||
# Try full-res first with q80
|
||||
img.save(output, "JPEG", quality=80, optimize=True)
|
||||
size_mb = output.stat().st_size / (1024 * 1024)
|
||||
|
||||
if size_mb > target_mb:
|
||||
w, h = img.size
|
||||
if w > max_width:
|
||||
h = int(h * max_width / w)
|
||||
w = max_width
|
||||
try:
|
||||
resample = Image.Resampling.LANCZOS
|
||||
except AttributeError:
|
||||
resample = Image.LANCZOS # type: ignore[attr-defined]
|
||||
img = img.resize((w, h), resample)
|
||||
img.save(output, "JPEG", quality=quality, optimize=True)
|
||||
size_mb = output.stat().st_size / (1024 * 1024)
|
||||
if size_mb > target_mb:
|
||||
img.save(output, "JPEG", quality=quality - 5, optimize=True)
|
||||
size_mb = output.stat().st_size / (1024 * 1024)
|
||||
|
||||
click.echo(f"Compressed: {img_file} -> {output} ({size_mb:.2f} MB)")
|
||||
@@ -0,0 +1,49 @@
|
||||
"""LaTeX utilities."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from bin.utils import handle_errors, run_command
|
||||
|
||||
|
||||
@click.group()
|
||||
def latex():
|
||||
"""LaTeX compilation and utilities."""
|
||||
pass
|
||||
|
||||
|
||||
@latex.command("compile")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--engine", "-e", default="pdflatex", help="LaTeX engine")
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def compile(files, engine, dry_run):
|
||||
"""Compile LaTeX files."""
|
||||
for pattern in files:
|
||||
for tex_file in Path(".").glob(pattern):
|
||||
if tex_file.suffix.lower() != ".tex":
|
||||
continue
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would compile: {tex_file}")
|
||||
continue
|
||||
|
||||
click.echo(f"Compiling: {tex_file}")
|
||||
run_command([engine, str(tex_file)], check=False)
|
||||
|
||||
|
||||
@latex.command("count")
|
||||
@click.argument("file", type=click.Path(exists=True))
|
||||
@handle_errors
|
||||
def count(file):
|
||||
"""Count words in a LaTeX file (excluding commands)."""
|
||||
result = run_command(
|
||||
["detex", file],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
words = len(result.stdout.split())
|
||||
click.echo(f"Words: {words}")
|
||||
@@ -0,0 +1,147 @@
|
||||
"""PDF utilities."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
from pypdf import PdfReader, PdfWriter
|
||||
from PIL import Image
|
||||
|
||||
from bin.utils import handle_errors, run_command
|
||||
|
||||
|
||||
@click.group()
|
||||
def pdf():
|
||||
"""PDF manipulation commands."""
|
||||
pass
|
||||
|
||||
|
||||
@pdf.command("compress")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option(
|
||||
"--quality",
|
||||
"-q",
|
||||
type=click.Choice(["screen", "ebook", "printer", "prepress"]),
|
||||
default="screen",
|
||||
help="Compression quality level",
|
||||
)
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def compress(files, quality, dry_run):
|
||||
"""Compress PDF files using ghostscript."""
|
||||
for pattern in files:
|
||||
for pdf_file in Path(".").glob(pattern):
|
||||
if pdf_file.suffix.lower() != ".pdf":
|
||||
continue
|
||||
|
||||
output = pdf_file.with_suffix(".compressed.pdf")
|
||||
cmd = [
|
||||
"gs",
|
||||
"-sDEVICE=pdfwrite",
|
||||
"-dNOPAUSE",
|
||||
"-dQUIET",
|
||||
"-dBATCH",
|
||||
f"-dPDFSETTINGS=/{quality}",
|
||||
"-dCompatibilityLevel=1.4",
|
||||
f"-sOutputFile={output}",
|
||||
str(pdf_file),
|
||||
]
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would compress: {pdf_file} -> {output}")
|
||||
continue
|
||||
|
||||
click.echo(f"Compressing: {pdf_file}")
|
||||
run_command(cmd, check=True)
|
||||
output.replace(pdf_file)
|
||||
click.echo(f" Done: {pdf_file}")
|
||||
|
||||
|
||||
@pdf.command("crop")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def crop(files, dry_run):
|
||||
"""Crop PDF files."""
|
||||
for pattern in files:
|
||||
for pdf_file in Path(".").glob(pattern):
|
||||
if pdf_file.suffix.lower() != ".pdf":
|
||||
continue
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would crop: {pdf_file}")
|
||||
continue
|
||||
|
||||
click.echo(f"Cropping: {pdf_file}")
|
||||
run_command(["pdfcrop", str(pdf_file), str(pdf_file)], check=False)
|
||||
|
||||
|
||||
@pdf.command("to-tiff")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--density", "-d", default="300", help="DPI density")
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def to_tiff(files, density, dry_run):
|
||||
"""Convert PDF to TIFF."""
|
||||
for pattern in files:
|
||||
for pdf_file in Path(".").glob(pattern):
|
||||
if pdf_file.suffix.lower() != ".pdf":
|
||||
continue
|
||||
output = pdf_file.with_suffix(".tiff")
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would convert: {pdf_file} -> {output}")
|
||||
continue
|
||||
|
||||
click.echo(f"Converting: {pdf_file} -> {output}")
|
||||
run_command(
|
||||
["convert", "-density", density, str(pdf_file), "-compress", "zip", str(output)],
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
@pdf.command("merge")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--output", "-o", required=True, help="Output PDF file")
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def merge_pdfs(files, output, dry_run):
|
||||
"""Merge PDFs and/or images into a single PDF."""
|
||||
writer = PdfWriter()
|
||||
processed = []
|
||||
|
||||
for pattern in files:
|
||||
for path in sorted(Path(".").glob(pattern)):
|
||||
if path.suffix.lower() == ".pdf":
|
||||
if dry_run:
|
||||
click.echo(f"Would add PDF: {path}")
|
||||
continue
|
||||
reader = PdfReader(str(path))
|
||||
for page in reader.pages:
|
||||
writer.add_page(page)
|
||||
processed.append(str(path))
|
||||
elif path.suffix.lower() in (".png", ".jpg", ".jpeg"):
|
||||
if dry_run:
|
||||
click.echo(f"Would add image: {path}")
|
||||
continue
|
||||
img = Image.open(path)
|
||||
if img.mode in ("RGBA", "P"):
|
||||
img = img.convert("RGB")
|
||||
# Save to temp PDF
|
||||
temp_pdf = path.with_suffix(".temp.pdf")
|
||||
img.save(temp_pdf, "PDF", resolution=150.0)
|
||||
reader = PdfReader(str(temp_pdf))
|
||||
writer.add_page(reader.pages[0])
|
||||
temp_pdf.unlink()
|
||||
processed.append(str(path))
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would write merged PDF to: {output}")
|
||||
return
|
||||
|
||||
if not processed:
|
||||
click.echo("No files found to merge")
|
||||
return
|
||||
|
||||
with open(output, "wb") as f:
|
||||
writer.write(f)
|
||||
click.echo(f"Merged {len(processed)} files into {output}")
|
||||
@@ -0,0 +1,69 @@
|
||||
"""Self-management commands for bin."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
|
||||
def _get_project_root() -> Path:
|
||||
"""Get mytoolkit project root."""
|
||||
# This file is at bin/commands/self_mgmt.py
|
||||
return Path(__file__).parent.parent.parent
|
||||
|
||||
|
||||
@click.command(name="update")
|
||||
def update_cmd():
|
||||
"""Update mytoolkit from the local repository."""
|
||||
root = _get_project_root()
|
||||
|
||||
click.secho("Updating mytoolkit...", fg="cyan")
|
||||
|
||||
# Run make install
|
||||
result = subprocess.run(
|
||||
["make", "install"],
|
||||
cwd=str(root),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
click.secho("mytoolkit updated successfully.", fg="green")
|
||||
else:
|
||||
click.secho("Update failed:", fg="red", err=True)
|
||||
click.echo(result.stderr, err=True)
|
||||
raise click.Exit(1)
|
||||
|
||||
|
||||
@click.command(name="uninstall")
|
||||
def uninstall_cmd():
|
||||
"""Uninstall mytoolkit."""
|
||||
root = _get_project_root()
|
||||
|
||||
click.secho("Uninstalling mytoolkit...", fg="cyan")
|
||||
|
||||
result = subprocess.run(
|
||||
["make", "uninstall"],
|
||||
cwd=str(root),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
click.secho("mytoolkit uninstalled.", fg="green")
|
||||
else:
|
||||
click.secho("Uninstall failed:", fg="red", err=True)
|
||||
click.echo(result.stderr, err=True)
|
||||
raise click.Exit(1)
|
||||
|
||||
|
||||
# Backward compatibility: keep self_cmd group for any existing scripts
|
||||
@click.group(name="self")
|
||||
def self_cmd():
|
||||
"""Manage mytoolkit itself (legacy, use 'update'/'uninstall' directly)."""
|
||||
pass
|
||||
|
||||
|
||||
# Register commands in the group for backward compatibility
|
||||
self_cmd.add_command(update_cmd, name="update")
|
||||
self_cmd.add_command(uninstall_cmd, name="uninstall")
|
||||
@@ -0,0 +1,111 @@
|
||||
"""SSH connection utilities."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from bin.config import config
|
||||
from bin.utils import handle_errors
|
||||
|
||||
|
||||
def _get_hosts() -> dict:
|
||||
"""Get SSH hosts from config."""
|
||||
hosts = {}
|
||||
for key in config.get_all().keys():
|
||||
if key.startswith("ssh_") and not key.endswith("_port") and not key.endswith("_key"):
|
||||
name = key[4:] # Remove 'ssh_' prefix
|
||||
host = config.get(f"ssh_{name}")
|
||||
port = config.get(f"ssh_{name}_port") or "22"
|
||||
key_file = config.get(f"ssh_{name}_key")
|
||||
if host:
|
||||
hosts[name] = {"host": host, "port": int(port)}
|
||||
if key_file:
|
||||
hosts[name]["key"] = key_file
|
||||
return hosts
|
||||
|
||||
|
||||
def _build_ssh_base_cmd(host_config: dict) -> list[str]:
|
||||
"""Build base SSH command with host and port."""
|
||||
cmd = ["ssh", "-p", str(host_config["port"])]
|
||||
|
||||
if "key" in host_config:
|
||||
key_path = Path(host_config["key"]).expanduser()
|
||||
cmd.extend(["-i", str(key_path)])
|
||||
|
||||
cmd.append(host_config["host"])
|
||||
return cmd
|
||||
|
||||
|
||||
@click.group(name="ssh")
|
||||
def ssh_cmd():
|
||||
"""SSH connection utilities."""
|
||||
pass
|
||||
|
||||
|
||||
@ssh_cmd.command("connect")
|
||||
@click.argument("name")
|
||||
@handle_errors
|
||||
def connect(name):
|
||||
"""Connect to a configured host."""
|
||||
hosts = _get_hosts()
|
||||
if name not in hosts:
|
||||
click.echo(f"Unknown host: {name}", err=True)
|
||||
click.echo(f"Available: {', '.join(hosts.keys())}", err=True)
|
||||
raise click.Exit(1)
|
||||
|
||||
cmd = _build_ssh_base_cmd(hosts[name])
|
||||
click.secho(f"Connecting to {name}...", fg="cyan")
|
||||
subprocess.run(cmd)
|
||||
|
||||
|
||||
@ssh_cmd.command("tunnel")
|
||||
@click.argument("name")
|
||||
@click.option("--local-port", "-l", default=11111, help="Local port")
|
||||
@click.option("--remote-port", "-r", default=11111, help="Remote port")
|
||||
@handle_errors
|
||||
def tunnel(name, local_port, remote_port):
|
||||
"""Create SSH tunnel with port forwarding."""
|
||||
hosts = _get_hosts()
|
||||
if name not in hosts:
|
||||
click.echo(f"Unknown host: {name}", err=True)
|
||||
raise click.Exit(1)
|
||||
|
||||
host_config = hosts[name]
|
||||
cmd = ["ssh", "-p", str(host_config["port"]), "-L", f"{local_port}:localhost:{remote_port}"]
|
||||
|
||||
if "key" in host_config:
|
||||
key_path = Path(host_config["key"]).expanduser()
|
||||
cmd.extend(["-i", str(key_path)])
|
||||
|
||||
cmd.append(host_config["host"])
|
||||
|
||||
click.secho(f"Creating tunnel {local_port} -> {remote_port} on {name}...", fg="cyan")
|
||||
click.secho(f"Command: {' '.join(cmd)}", fg="dim")
|
||||
subprocess.run(cmd)
|
||||
|
||||
|
||||
@ssh_cmd.command("list")
|
||||
def list_hosts():
|
||||
"""List configured SSH hosts."""
|
||||
hosts = _get_hosts()
|
||||
click.secho("Configured hosts:", fg="cyan")
|
||||
for name, host_config in sorted(hosts.items()):
|
||||
click.echo(f" {name:12} {host_config['host']}:{host_config['port']}")
|
||||
|
||||
|
||||
@ssh_cmd.command("cmd")
|
||||
@click.argument("name")
|
||||
@click.argument("command")
|
||||
@handle_errors
|
||||
def run_cmd(name, command):
|
||||
"""Run a command on remote host."""
|
||||
hosts = _get_hosts()
|
||||
if name not in hosts:
|
||||
click.echo(f"Unknown host: {name}", err=True)
|
||||
raise click.Exit(1)
|
||||
|
||||
cmd = _build_ssh_base_cmd(hosts[name])
|
||||
cmd.append(command)
|
||||
|
||||
subprocess.run(cmd)
|
||||
@@ -0,0 +1,32 @@
|
||||
"""Utility commands."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from bin.utils import handle_errors
|
||||
|
||||
|
||||
@click.group()
|
||||
def utils():
|
||||
"""Miscellaneous utilities."""
|
||||
pass
|
||||
|
||||
|
||||
@utils.command("albany-cleanup")
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def albany_cleanup(dry_run):
|
||||
"""Clean up Albany temporary files (phalanx_*)."""
|
||||
count = 0
|
||||
for file in Path(".").glob("phalanx_*"):
|
||||
if dry_run:
|
||||
click.echo(f"Would remove: {file}")
|
||||
else:
|
||||
file.unlink()
|
||||
count += 1
|
||||
|
||||
if dry_run:
|
||||
click.echo("Dry run completed")
|
||||
else:
|
||||
click.echo(f"Removed {count} Albany temporary files")
|
||||
@@ -0,0 +1,37 @@
|
||||
"""Video utilities."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from bin.utils import handle_errors, run_command
|
||||
|
||||
|
||||
@click.group()
|
||||
def video():
|
||||
"""Video conversion commands."""
|
||||
pass
|
||||
|
||||
|
||||
@video.command("avi-to-mp4")
|
||||
@click.argument("files", nargs=-1, required=True)
|
||||
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done")
|
||||
@handle_errors
|
||||
def avi_to_mp4(files, dry_run):
|
||||
"""Convert AVI files to MP4 using ffmpeg."""
|
||||
for pattern in files:
|
||||
for avi_file in Path(".").glob(pattern):
|
||||
if avi_file.suffix.lower() != ".avi":
|
||||
continue
|
||||
output = avi_file.with_suffix(".mp4")
|
||||
|
||||
if dry_run:
|
||||
click.echo(f"Would convert: {avi_file} -> {output}")
|
||||
continue
|
||||
|
||||
click.echo(f"Converting: {avi_file} -> {output}")
|
||||
run_command(
|
||||
["ffmpeg", "-i", str(avi_file), "-c:v", "libx264", "-c:a", "aac", str(output)],
|
||||
check=False,
|
||||
)
|
||||
Reference in New Issue
Block a user