297 lines
10 KiB
Python
Executable File
297 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Print Queue Manager - Store items to print when you're not on your home network
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import tempfile
|
|
|
|
QUEUE_FILE = "queue.json"
|
|
WATCH_FOLDER = Path.home() / "to-print"
|
|
PRINTED_FOLDER = WATCH_FOLDER / "printed"
|
|
|
|
|
|
class PrintQueue:
|
|
def __init__(self):
|
|
self.queue_file = Path(__file__).parent / QUEUE_FILE
|
|
self.queue = self.load_queue()
|
|
self.watch_folder = WATCH_FOLDER
|
|
self.printed_folder = PRINTED_FOLDER
|
|
|
|
# Create watch folder if it doesn't exist
|
|
self.watch_folder.mkdir(exist_ok=True)
|
|
self.printed_folder.mkdir(exist_ok=True)
|
|
|
|
def load_queue(self):
|
|
"""Load the print queue from JSON file"""
|
|
if self.queue_file.exists():
|
|
with open(self.queue_file, 'r') as f:
|
|
return json.load(f)
|
|
return []
|
|
|
|
def save_queue(self):
|
|
"""Save the print queue to JSON file"""
|
|
with open(self.queue_file, 'w') as f:
|
|
json.dump(self.queue, f, indent=2)
|
|
|
|
def add(self, item, item_type, name=None):
|
|
"""Add an item to the print queue"""
|
|
entry = {
|
|
"id": len(self.queue) + 1,
|
|
"type": item_type,
|
|
"item": item,
|
|
"name": name or item,
|
|
"added": datetime.now().isoformat(),
|
|
"printed": False
|
|
}
|
|
self.queue.append(entry)
|
|
self.save_queue()
|
|
print(f"✓ Added to queue: {entry['name']}")
|
|
return entry
|
|
|
|
def list_items(self, show_all=False):
|
|
"""List items in the queue"""
|
|
items = self.queue if show_all else [i for i in self.queue if not i['printed']]
|
|
folder_files = self.get_folder_files()
|
|
|
|
if not items and not folder_files:
|
|
print("Nothing to print!")
|
|
return
|
|
|
|
if items:
|
|
print("\nManually Added Queue:")
|
|
print("-" * 60)
|
|
for item in items:
|
|
status = "✓ PRINTED" if item['printed'] else "⏳ PENDING"
|
|
print(f"[{item['id']}] {status}")
|
|
print(f" Name: {item['name']}")
|
|
print(f" Type: {item['type']}")
|
|
print(f" Added: {item['added'][:10]}")
|
|
if item['printed']:
|
|
print(f" Printed: {item.get('printed_at', 'N/A')[:10]}")
|
|
print()
|
|
|
|
if folder_files:
|
|
print(f"\nFiles in Watch Folder ({self.watch_folder}):")
|
|
print("-" * 60)
|
|
for file in folder_files:
|
|
print(f"⏳ {file.name}")
|
|
print(f"\nTotal: {len(folder_files)} file(s)")
|
|
print()
|
|
|
|
def print_all(self, dry_run=False):
|
|
"""Print all pending items in the queue and folder"""
|
|
pending = [i for i in self.queue if not i['printed']]
|
|
folder_files = self.get_folder_files()
|
|
|
|
total = len(pending) + len(folder_files)
|
|
|
|
if total == 0:
|
|
print("Nothing to print!")
|
|
return
|
|
|
|
print(f"\n{'Would print' if dry_run else 'Printing'} {total} item(s)...\n")
|
|
|
|
# Print queue items
|
|
if pending:
|
|
print("From manually added queue:")
|
|
for item in pending:
|
|
try:
|
|
if dry_run:
|
|
print(f" Would print: {item['name']} ({item['type']})")
|
|
else:
|
|
print(f" Printing: {item['name']}...")
|
|
self.print_item(item)
|
|
item['printed'] = True
|
|
item['printed_at'] = datetime.now().isoformat()
|
|
print(f" ✓ Success")
|
|
except Exception as e:
|
|
print(f" ✗ Failed: {e}")
|
|
|
|
if not dry_run:
|
|
self.save_queue()
|
|
print()
|
|
|
|
# Print folder files
|
|
if folder_files:
|
|
print(f"From watch folder ({self.watch_folder}):")
|
|
for file in folder_files:
|
|
try:
|
|
if dry_run:
|
|
print(f" Would print: {file.name}")
|
|
else:
|
|
print(f" Printing: {file.name}...")
|
|
self.print_folder_file(file)
|
|
print(f" ✓ Success (moved to {self.printed_folder.name}/)")
|
|
except Exception as e:
|
|
print(f" ✗ Failed: {e}")
|
|
print()
|
|
|
|
print("Done! Use 'list --all' to see all items or 'clear' to remove printed items.")
|
|
|
|
def print_item(self, item):
|
|
"""Print a single item"""
|
|
if item['type'] == 'file':
|
|
# Expand user home directory and make path absolute
|
|
file_path = Path(item['item']).expanduser().resolve()
|
|
if not file_path.exists():
|
|
raise FileNotFoundError(f"File not found: {file_path}")
|
|
subprocess.run(['lp', str(file_path)], check=True, capture_output=True)
|
|
|
|
elif item['type'] == 'url':
|
|
# Download URL and print
|
|
# For web pages, you might want to use wkhtmltopdf for better formatting
|
|
with tempfile.NamedTemporaryFile(suffix='.html', delete=False) as tmp:
|
|
try:
|
|
print(f" Downloading...")
|
|
result = subprocess.run(
|
|
['curl', '-L', '-s', item['item'], '-o', tmp.name],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
|
|
# Check if file has content
|
|
if os.path.getsize(tmp.name) == 0:
|
|
raise ValueError("Downloaded file is empty")
|
|
|
|
print(f" Sending to printer...")
|
|
subprocess.run(['lp', tmp.name], check=True, capture_output=True)
|
|
finally:
|
|
# Clean up temp file
|
|
if os.path.exists(tmp.name):
|
|
os.unlink(tmp.name)
|
|
|
|
def clear_printed(self):
|
|
"""Remove printed items from the queue"""
|
|
original_count = len(self.queue)
|
|
self.queue = [i for i in self.queue if not i['printed']]
|
|
removed = original_count - len(self.queue)
|
|
self.save_queue()
|
|
print(f"✓ Removed {removed} printed item(s) from queue")
|
|
|
|
def remove_item(self, item_id):
|
|
"""Remove a specific item from the queue by ID"""
|
|
original_queue = self.queue.copy()
|
|
self.queue = [i for i in self.queue if i['id'] != item_id]
|
|
|
|
if len(self.queue) == len(original_queue):
|
|
print(f"✗ Item {item_id} not found")
|
|
return False
|
|
|
|
self.save_queue()
|
|
print(f"✓ Removed item {item_id} from queue")
|
|
return True
|
|
|
|
def get_folder_files(self):
|
|
"""Get all printable files from the watch folder"""
|
|
# Common printable file extensions
|
|
printable_extensions = {'.pdf', '.txt', '.jpg', '.jpeg', '.png', '.gif',
|
|
'.doc', '.docx', '.html', '.htm', '.ps'}
|
|
|
|
files = []
|
|
if self.watch_folder.exists():
|
|
for file in self.watch_folder.iterdir():
|
|
if file.is_file() and file.suffix.lower() in printable_extensions:
|
|
files.append(file)
|
|
|
|
return sorted(files, key=lambda f: f.stat().st_mtime)
|
|
|
|
def print_folder_file(self, file_path):
|
|
"""Print a file from the watch folder and move it to printed folder"""
|
|
subprocess.run(['lp', str(file_path)], check=True, capture_output=True)
|
|
|
|
# Move to printed folder
|
|
dest = self.printed_folder / file_path.name
|
|
# Handle name conflicts
|
|
counter = 1
|
|
while dest.exists():
|
|
stem = file_path.stem
|
|
suffix = file_path.suffix
|
|
dest = self.printed_folder / f"{stem}_{counter}{suffix}"
|
|
counter += 1
|
|
|
|
file_path.rename(dest)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description='Print Queue Manager - Store and print files when on your home network',
|
|
epilog='Examples:\n'
|
|
' %(prog)s add document.pdf\n'
|
|
' %(prog)s add https://example.com/page.html --name "Example Page"\n'
|
|
' %(prog)s list\n'
|
|
' %(prog)s print\n',
|
|
formatter_class=argparse.RawDescriptionHelpFormatter
|
|
)
|
|
|
|
subparsers = parser.add_subparsers(dest='command', help='Available commands')
|
|
|
|
# Add command
|
|
add_parser = subparsers.add_parser('add', help='Add item to print queue')
|
|
add_parser.add_argument('item', help='File path or URL to print')
|
|
add_parser.add_argument('--name', help='Custom name for the item')
|
|
|
|
# List command
|
|
list_parser = subparsers.add_parser('list', help='List items in queue')
|
|
list_parser.add_argument('--all', action='store_true',
|
|
help='Show all items including printed ones')
|
|
|
|
# Print command
|
|
print_parser = subparsers.add_parser('print', help='Print all pending items')
|
|
print_parser.add_argument('--dry-run', action='store_true',
|
|
help='Show what would be printed without actually printing')
|
|
|
|
# Clear command
|
|
clear_parser = subparsers.add_parser('clear',
|
|
help='Remove printed items from queue')
|
|
|
|
# Remove command
|
|
remove_parser = subparsers.add_parser('remove',
|
|
help='Remove a specific item by ID')
|
|
remove_parser.add_argument('id', type=int, help='Item ID to remove')
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
return
|
|
|
|
queue = PrintQueue()
|
|
|
|
if args.command == 'add':
|
|
# Detect if it's a URL or file
|
|
item_type = 'url' if args.item.startswith(('http://', 'https://')) else 'file'
|
|
|
|
# Validate file exists if it's a file path
|
|
if item_type == 'file':
|
|
file_path = Path(args.item).expanduser().resolve()
|
|
if not file_path.exists():
|
|
print(f"✗ Error: File not found: {file_path}")
|
|
sys.exit(1)
|
|
# Store the absolute path
|
|
args.item = str(file_path)
|
|
|
|
queue.add(args.item, item_type, args.name)
|
|
|
|
elif args.command == 'list':
|
|
queue.list_items(show_all=args.all)
|
|
|
|
elif args.command == 'print':
|
|
queue.print_all(dry_run=args.dry_run)
|
|
|
|
elif args.command == 'clear':
|
|
queue.clear_printed()
|
|
|
|
elif args.command == 'remove':
|
|
queue.remove_item(args.id)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|