374 lines
13 KiB
Python
374 lines
13 KiB
Python
"""
|
|
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() |