""" 统一服务器 - 合并H5页面 + OCR + 导出服务于一个端口 手机端通过内网穿透访问 """ import json import io import os import sys import re import base64 import tempfile from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse # 确保能导入同目录模块 WORK_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, WORK_DIR) # 导入导出服务核心函数 from export_server import generate_docx # 导入OCR核心函数 import pytesseract import platform import shutil from PIL import Image # 跨平台 Tesseract 路径配置 if platform.system() == 'Windows': # Windows 本地开发环境 pytesseract.pytesseract.tesseract_cmd = r'C:\Program Files\Tesseract-OCR\tesseract.exe' os.environ['TESSDATA_PREFIX'] = os.path.expandvars(r'%USERPROFILE%\.tesseract\tessdata') else: # Linux/Docker 环境：tesseract 已通过 apt 安装到 PATH tesseract_bin = shutil.which('tesseract') if tesseract_bin: pytesseract.pytesseract.tesseract_cmd = tesseract_bin # 自动检测 tessdata 路径（兼容 Tesseract 4.x/5.x） for _tessdata in ['/usr/share/tesseract-ocr/5/tessdata', '/usr/share/tesseract-ocr/4.00/tessdata', '/usr/share/tessdata']: if os.path.isdir(_tessdata): os.environ.setdefault('TESSDATA_PREFIX', _tessdata) break print(f'[Tesseract] cmd={pytesseract.pytesseract.tesseract_cmd}, ' f'TESSDATA_PREFIX={os.environ.get("TESSDATA_PREFIX", "(default)")}') # H5页面路径 H5_FILE = os.path.join(WORK_DIR, 'test-h5.html') PORT = int(os.environ.get('PORT', 7860)) def ocr_image(image_data): """对图片进行OCR识别""" img = Image.open(io.BytesIO(image_data)) w, h = img.size max_dim = 2000 if w > max_dim or h > max_dim: ratio = min(max_dim / w, max_dim / h) img = img.resize((int(w * ratio), int(h * ratio)), Image.LANCZOS) text = pytesseract.image_to_string(img, lang='chi_sim', config='--psm 6') return text.strip() class UnifiedHandler(BaseHTTPRequestHandler): def _add_cors_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') def do_OPTIONS(self): self.send_response(200) self._add_cors_headers() self.end_headers() def do_GET(self): """处理GET请求 - 返回H5页面""" path = urlparse(self.path).path if path == '/' or path == '/test-h5.html' or path == '/index.html': try: with open(H5_FILE, 'r', encoding='utf-8') as f: content = f.read() self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self._add_cors_headers() self.end_headers() self.wfile.write(content.encode('utf-8')) except Exception as e: self.send_error(500, f'读取页面失败: {e}') elif path == '/health': # 健康检查 self.send_response(200) self.send_header('Content-Type', 'application/json') self._add_cors_headers() self.end_headers() self.wfile.write(json.dumps({'status': 'ok'}).encode('utf-8')) else: self.send_error(404, '页面不存在') def do_POST(self): """处理POST请求 - OCR和导出""" path = urlparse(self.path).path if path == '/ocr': self._handle_ocr() elif path == '/export': self._handle_export() else: self.send_error(404, '接口不存在') def _handle_ocr(self): """处理OCR识别请求""" content_length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(content_length) content_type = self.headers.get('Content-Type', '') if 'multipart/form-data' not in content_type: self._send_json({'error': '需要 multipart/form-data'}, 400) return # 提取 boundary boundary = content_type.split('boundary=')[1].strip() if boundary.startswith('"') and boundary.endswith('"'): boundary = boundary[1:-1] # 手动解析 multipart parts = body.split(('--' + boundary).encode()) image_data = None for part in parts: if b'Content-Disposition' in part and b'filename=' in part: header_end = part.find(b'\r\n\r\n') if header_end == -1: continue file_data = part[header_end + 4:] if file_data.endswith(b'\r\n'): file_data = file_data[:-2] image_data = file_data break if not image_data: self._send_json({'error': '未找到图片数据'}, 400) return try: text = ocr_image(image_data) self._send_json({'text': text, 'length': len(text)}) print(f'[OCR] 识别成功，提取 {len(text)} 个字符') except Exception as e: self._send_json({'error': f'OCR识别失败: {str(e)}'}, 500) print(f'[OCR] 识别失败: {e}') def _handle_export(self): """处理Word文档导出请求""" content_length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(content_length) try: visit_data = json.loads(body) buffer = generate_docx(visit_data) self.send_response(200) self.send_header('Content-Type', 'application/vnd.openxmlformats-officedocument.wordprocessingml.document') self._add_cors_headers() filename = 'visit_%s_%sstores.docx' % ( visit_data.get('visitDateDisplay', '').replace('.', '-'), visit_data.get('totalCount', 0) ) self.send_header('Content-Disposition', f'attachment; filename="{filename}"') self.end_headers() self.wfile.write(buffer.getvalue()) print(f'[Export] 导出成功: {filename}') except Exception as e: self.send_response(500) self.send_header('Content-Type', 'application/json') self._add_cors_headers() self.end_headers() self.wfile.write(json.dumps({'error': str(e)}).encode('utf-8')) print(f'[Export] 导出失败: {e}') def _send_json(self, data, status=200): self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self._add_cors_headers() self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8')) def log_message(self, format, *args): print(f'[{self.log_date_time_string()}] {args[0]}') if __name__ == '__main__': print('=' * 50) print(' 药店拜访助手 - 统一服务器') print(f' 本机访问: http://127.0.0.1:{PORT}') print(f' 局域网访问: http://<本机IP>:{PORT}') print('=' * 50) # 绑定 0.0.0.0 让局域网设备也能访问 server = HTTPServer(('0.0.0.0', PORT), UnifiedHandler) try: server.serve_forever() except KeyboardInterrupt: print('\n服务器已停止') server.shutdown()