updated to silent print
This commit is contained in:
@@ -1,374 +0,0 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user