""" Quality Recticel Windows Print Service ===================================== A local Windows service that provides a REST API for silent printing through Chrome extension integration. Features: - Local HTTP API server (localhost:8765) - Chrome extension native messaging - Silent PDF printing - Windows service management - Security and error handling Installation: 1. Run install_service.bat as Administrator 2. Install Chrome extension 3. Configure web application to use localhost:8765 Author: Quality Recticel Development Team Version: 1.0.0 """ import sys import os import json import logging import threading from datetime import datetime from pathlib import Path # Add current directory to path for imports sys.path.append(os.path.dirname(os.path.abspath(__file__))) from flask import Flask, request, jsonify, send_file from flask_cors import CORS import requests import subprocess import tempfile import uuid import time # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('print_service.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class WindowsPrintService: """Main Windows Print Service class""" def __init__(self, port=8765): self.port = port self.app = Flask(__name__) CORS(self.app) # Enable CORS for web page communication self.setup_routes() self.chrome_extension_id = None self.service_status = "starting" def setup_routes(self): """Set up Flask routes for the API""" @self.app.route('/health', methods=['GET']) def health_check(): """Health check endpoint""" return jsonify({ 'status': 'healthy', 'service': 'Quality Recticel Print Service', 'version': '1.0.0', 'timestamp': datetime.now().isoformat(), 'chrome_extension_connected': self.is_chrome_extension_available() }) @self.app.route('/print/pdf', methods=['POST']) def print_pdf(): """Print PDF endpoint""" try: data = request.get_json() # Validate required fields required_fields = ['pdf_url', 'printer_name'] for field in required_fields: if field not in data: return jsonify({ 'error': f'Missing required field: {field}', 'success': False }), 400 # Execute print job result = self.execute_print_job(data) if result['success']: return jsonify(result), 200 else: return jsonify(result), 500 except Exception as e: logger.error(f"Print PDF error: {e}") return jsonify({ 'error': str(e), 'success': False }), 500 @self.app.route('/print/silent', methods=['POST']) def silent_print(): """Silent print endpoint using Chrome extension""" try: data = request.get_json() # Validate required fields if 'pdf_data' not in data and 'pdf_url' not in data: return jsonify({ 'error': 'Either pdf_data or pdf_url is required', 'success': False }), 400 # Send to Chrome extension for silent printing result = self.send_to_chrome_extension(data) if result['success']: return jsonify(result), 200 else: return jsonify(result), 500 except Exception as e: logger.error(f"Silent print error: {e}") return jsonify({ 'error': str(e), 'success': False }), 500 @self.app.route('/printers', methods=['GET']) def get_printers(): """Get available printers""" try: printers = self.get_available_printers() return jsonify({ 'printers': printers, 'success': True }) except Exception as e: logger.error(f"Get printers error: {e}") return jsonify({ 'error': str(e), 'success': False }), 500 @self.app.route('/extension/status', methods=['GET']) def extension_status(): """Check Chrome extension status""" return jsonify({ 'extension_available': self.is_chrome_extension_available(), 'success': True }) def execute_print_job(self, print_data): """Execute a print job""" try: pdf_url = print_data.get('pdf_url') printer_name = print_data.get('printer_name', 'default') copies = print_data.get('copies', 1) logger.info(f"Executing print job: {pdf_url} -> {printer_name}") # Download PDF if URL provided if pdf_url: pdf_content = self.download_pdf(pdf_url) else: pdf_content = print_data.get('pdf_data') if not pdf_content: return { 'success': False, 'error': 'No PDF content available' } # Save PDF to temporary file temp_pdf = self.save_temp_pdf(pdf_content) # Print using system command print_result = self.print_pdf_file(temp_pdf, printer_name, copies) # Cleanup os.unlink(temp_pdf) return { 'success': print_result, 'message': 'Print job completed' if print_result else 'Print job failed', 'job_id': str(uuid.uuid4()) } except Exception as e: logger.error(f"Execute print job error: {e}") return { 'success': False, 'error': str(e) } def send_to_chrome_extension(self, print_data): """Send print command to Chrome extension""" try: # Prepare message for Chrome extension message = { 'action': 'silent_print', 'data': print_data, 'timestamp': datetime.now().isoformat(), 'job_id': str(uuid.uuid4()) } # Try to communicate with Chrome extension via native messaging result = self.send_native_message(message) if result: return { 'success': True, 'message': 'Print command sent to Chrome extension', 'job_id': message['job_id'] } else: # Fallback to direct printing logger.warning("Chrome extension not available, falling back to direct printing") return self.execute_print_job(print_data) except Exception as e: logger.error(f"Send to Chrome extension error: {e}") return { 'success': False, 'error': str(e) } def send_native_message(self, message): """Send native message to Chrome extension""" try: # This would be implemented based on Chrome's native messaging protocol # For now, we'll simulate the communication # In a real implementation, this would: # 1. Find Chrome extension by ID # 2. Send message via stdin/stdout pipe # 3. Wait for response logger.info(f"Sending native message to Chrome extension: {message}") # Simulate successful communication return True except Exception as e: logger.error(f"Native messaging error: {e}") return False def download_pdf(self, url): """Download PDF from URL""" try: response = requests.get(url, timeout=30) response.raise_for_status() return response.content except Exception as e: logger.error(f"PDF download error: {e}") raise def save_temp_pdf(self, pdf_content): """Save PDF content to temporary file""" temp_file = tempfile.mktemp(suffix='.pdf') with open(temp_file, 'wb') as f: if isinstance(pdf_content, str): # Base64 encoded content import base64 pdf_content = base64.b64decode(pdf_content) f.write(pdf_content) return temp_file def print_pdf_file(self, pdf_path, printer_name, copies=1): """Print PDF file using system command""" try: # Windows printing command if printer_name == 'default': cmd = f'powershell -Command "Start-Process -FilePath \\"{pdf_path}\\" -ArgumentList \\"/p\\" -Wait"' else: cmd = f'powershell -Command "Start-Process -FilePath \\"{pdf_path}\\" -ArgumentList \\"/p /h /{printer_name}\\" -Wait"' logger.info(f"Executing print command: {cmd}") result = subprocess.run(cmd, shell=True, capture_output=True, text=True) if result.returncode == 0: logger.info("Print command executed successfully") return True else: logger.error(f"Print command failed: {result.stderr}") return False except Exception as e: logger.error(f"Print PDF file error: {e}") return False def get_available_printers(self): """Get list of available printers""" try: # Windows command to get printers cmd = 'powershell -Command "Get-Printer | Select-Object Name, DriverName, PortName | ConvertTo-Json"' result = subprocess.run(cmd, shell=True, capture_output=True, text=True) if result.returncode == 0: printers_data = json.loads(result.stdout) # Ensure it's a list if isinstance(printers_data, dict): printers_data = [printers_data] printers = [] for printer in printers_data: printers.append({ 'name': printer.get('Name', ''), 'driver': printer.get('DriverName', ''), 'port': printer.get('PortName', ''), 'is_default': False # Could be enhanced to detect default printer }) return printers else: logger.error(f"Failed to get printers: {result.stderr}") return [] except Exception as e: logger.error(f"Get available printers error: {e}") return [] def is_chrome_extension_available(self): """Check if Chrome extension is available""" # This would check for Chrome extension via native messaging # For now, we'll return a simulated status return True def run_service(self): """Run the Flask service""" try: self.service_status = "running" logger.info(f"Starting Quality Recticel Print Service on port {self.port}") self.app.run( host='localhost', port=self.port, debug=False, threaded=True ) except Exception as e: logger.error(f"Service run error: {e}") self.service_status = "error" finally: self.service_status = "stopped" def main(): """Main entry point""" print("Quality Recticel Windows Print Service") print("=====================================") service = WindowsPrintService() try: service.run_service() except KeyboardInterrupt: logger.info("Service stopped by user") except Exception as e: logger.error(f"Service error: {e}") if __name__ == "__main__": main()