import os import re import shutil import logging import subprocess import json from datetime import datetime import sys # 配置日志 log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'log') os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, f"eric_run_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log") logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file), logging.StreamHandler() ] ) logger = logging.getLogger() # 设置脚本所在目录为工作目录 script_dir = os.path.dirname(os.path.abspath(__file__)) os.chdir(script_dir) work_dir = script_dir # 环境配置（根据实际情况修改） env_config = { 'test_cert_file': 'test-cert.pfx', 'test_cert_pass': '123456', 'prod_cert_file': 'prod-cert.pfx', 'prod_cert_pass': 'Meiouwang1332!' } def eric_run(xml_path: str, declare_type: str, is_test: int = 0, tid: str = ""): """ 执行ERIC申报并处理结果 Args: xml_path: XML文件路径 declare_type: 申报类型 is_test: 是否为测试模式（0=正式，1=测试） tid: 交易ID Returns: dict: {"success": 成功标志, "pdf_path": PDF文件路径, "return_info": 返回信息} """ flag = False python_executable = sys.executable # 获取当前 Python 解释器路径 if is_test: cert_file = os.path.join(work_dir, 'cert', env_config['test_cert_file']) passwd = f"{env_config['test_cert_pass']}" else: cert_file = os.path.join(work_dir, 'cert', env_config['prod_cert_file']) passwd = f"{env_config['prod_cert_pass']}" logger.info(f"开始执行ERIC申报: 类型={declare_type}, XML路径={xml_path}, 测试模式={is_test}, TID={tid}") logger.info(f"使用证书: {cert_file}") # 切换到eric目录执行，因为ericprint.pdf可能生成在那里 eric_dir = os.path.join(work_dir, 'eric') command = [ sys.executable, os.path.join(work_dir, 'eric', 'ericdemo.py'), "/v", declare_type, "/x", xml_path, "/c", cert_file, "/p", passwd, "/d", os.path.join(work_dir, 'lib'), "/l", os.path.join(work_dir, 'log'), "/s", os.path.join(work_dir, 'response'), "/senden" ] logger.info(f"执行命令: {' '.join(command)}") try: # 保存当前工作目录 current_dir = os.getcwd() # 切换到eric目录执行 os.chdir(eric_dir) result = subprocess.run( command, capture_output=True, text=True, input='\n', timeout=60 ) # 切换回原工作目录 os.chdir(current_dir) result_text = result.stdout dest_pdf_name = '' # 记录命令输出 logger.info("=== 命令输出 ===") logger.info(result_text) if result.stderr: logger.error("=== 错误信息 ===") logger.error(result.stderr) logger.info(f"命令执行完成，返回码: {result.returncode}") # 扩展的错误结果匹配：覆盖更多 ERIC 错误消息模式 error_patterns = [ r'Datensatz\s+nicht\s+plausibel', r'Fehler\s+beim\s+Validieren', r'Validierung\s+des\s+XML', r'Datenartversion\s+ist\s+unbekannt', r'ERiC-Plugin\s+wurde\s+nicht\s+gefunden', r'Fehler\s+beim\s+Versenden', r'Sendestatus:(?!\s*Daten\s+wurden\s+erfolgreich\s+angenommen)(?!\s*Verarbeitung\s+fehlerfrei).*\b(fehler|Fehler|unbekannt|nicht\s+gefunden|nicht\s+plausibel)\b', ] has_error = False matched_patterns = [] for pat in error_patterns: m = re.search(pat, result_text, re.IGNORECASE) if m: has_error = True matched_patterns.append(pat) logger.error(f"检测到 ERIC 错误: pattern={pat}") # 从 eric.log 提取详细错误信息（ERIC 说 "Details stehen im Logfile" 时特别有用） eric_log_details = "" try: import glob as _glob log_dir_py = os.path.join(work_dir, 'log') if os.path.isdir(log_dir_py): # 找最近的 eric-*.log 文件 log_files = sorted( _glob.glob(os.path.join(log_dir_py, 'eric*.log')) + _glob.glob(os.path.join(log_dir_py, '*eric*.log')), key=lambda p: os.path.getmtime(p), reverse=True ) if log_files: latest = log_files[0] try: with open(latest, 'r', encoding='utf-8', errors='replace') as fh: lines = fh.readlines() # 只保留最近 100 行，过滤出 ERROR / Fehler / ungueltig 等关键行 recent_lines = lines[-100:] if len(lines) > 100 else lines interesting = [] for l in recent_lines: ll = l.strip() if not ll: continue lc = ll.lower() if any(k in lc for k in ['error', 'fehler', 'ungueltig', 'ungültig', 'unbekannt', 'nicht gefunden', 'validieren', 'validierung', 'schema', 'required', 'fehlt', 'missing', 'datensatz']): interesting.append(ll) elif 'E30' in ll or 'E31' in ll: # 可能是字段名 interesting.append(ll) # 如果没找到关键行，至少返回最后 30 行 if not interesting: interesting = [l.strip() for l in recent_lines[-30:] if l.strip()] if interesting: eric_log_details = "\n[eric.log 最近相关内容]\n" + "\n".join(interesting[-30:]) logger.info("从 eric.log 提取到 " + str(len(interesting)) + " 行相关内容") except Exception as _ex: logger.warning(f"读取 eric.log 失败: {_ex}") except Exception as _e: logger.warning(f"扫描 eric.log 目录失败: {_e}") # 是否申报成功字样匹配 success_matches = re.findall(r'Verarbeitung\s+fehlerfrei|Daten wurden erfolgreich angenommen|Code>0 0 else result_text if eric_log_details: return_info = return_info + eric_log_details print(success_matches) if has_error: # 检测到错误模式，申报出错 logger.error(f"ERIC 检测到错误模式: {matched_patterns}, return_info={return_info[:500]}") return {"success": False, "pdf_path": "", "return_info": return_info} elif len(success_matches): # 需要判断是否有: Serverantwort 字样 # 申报成功, 将pdf改名且复制到新的目录 # 检查eric目录下是否有ericprint.pdf文件 ericprint_path = os.path.join(eric_dir, 'ericprint.pdf') work_dir_pdf_path = os.path.join(work_dir, 'ericprint.pdf') # 即使没有PDF文件，只要服务器返回成功，也认为申报成功 xml_matches = re.findall(r"Serverantwort\s*:[\n]*(<\?xml.*?)", _rtext, re.DOTALL | re.IGNORECASE) return_info = xml_matches[0].strip() if xml_matches else "" flag = True # 确保temp目录存在 temp_dir = os.path.join(work_dir, 'temp') if not os.path.exists(temp_dir): os.makedirs(temp_dir, exist_ok=True) logger.info(f"创建temp目录: {temp_dir}") if os.path.isfile(ericprint_path): temp_pdf_name = f"{tid}_{declare_type}.pdf" temp_pdf_path = os.path.join(work_dir, 'temp', temp_pdf_name) # 复制文件到work_dir目录 shutil.copy2(ericprint_path, temp_pdf_path) logger.info(f"PDF文件已从eric目录复制到work_dir目录: {temp_pdf_path}") # 如果提供了tid参数，复制到log目录 if tid: # 构建目标文件名 log_pdf_name = f"{tid}_{declare_type}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf" log_pdf_path = os.path.join(log_dir, log_pdf_name) # 复制文件到log目录 shutil.copy2(ericprint_path, log_pdf_path) logger.info(f"PDF文件已复制到log目录: {log_pdf_path}") # 返回temp目录中的文件路径 return_pdf_path = temp_pdf_path else: logger.info("申报成功，但未生成PDF文件") return_pdf_path = "" return {"success": flag, "pdf_path": return_pdf_path, "return_info": return_info} logger.warning(f"申报结果未明确: {return_info}") return {"success": flag, "pdf_path": dest_pdf_name, "return_info": return_info} except Exception as e: # 确保切换回原工作目录 try: os.chdir(current_dir) except: pass error_msg = f"执行ERIC申报时出错: {str(e)}" logger.error(error_msg) return {"success": False, "pdf_path": "", "return_info": error_msg} def parse_declare_type(xml_basename: str) -> str: """ 从 XML 文件名 (不含扩展名) 解析 ERIC 需要的 declare_type / DatenArtVersion 示例: "USt_2025" -> "USt_2025" "UStVA_2026" -> "UStVA_2026" "ZMDO_2025" -> "ZMDO" "USt_2025_POVAT..." -> "USt_2025" "UStVA_2026_POVAT..." -> "UStVA_2026" """ # 1) ZMDO 不分年份 if 'ZMDO' in xml_basename.upper(): return 'ZMDO' # 2) 使用正则匹配 前缀_YYYY 格式，忽略后面 _TID 等多余内容 import re as _re m = _re.match(r'^(UStVA|USt)_(\d{4})', xml_basename, _re.IGNORECASE) if m: prefix = m.group(1) year = m.group(2) # 标准化前缀大小写 if prefix.upper().startswith('USTVA'): return f'UStVA_{year}' else: return f'USt_{year}' # 3) 兜底：直接返回原文件全名（可能不是我们设计的文件名格式） logger.warning(f"无法从文件名 '%s' 无法解析出标准 declare_type，返回原名", xml_basename) return xml_basename if __name__ == "__main__": # 获取命令行参数 if len(sys.argv) < 4: error_msg = "错误：缺少必要的命令行参数" logger.error(error_msg) print(json.dumps({"success": False, "pdf_path": "", "return_info": error_msg})) sys.exit(1) try: # 检查是否是直接传递XML路径的调用方式 if os.path.isfile(sys.argv[1]): # 从PHP调用的方式：python run_ericdemo_year.py xml_path = sys.argv[1] output_dir = sys.argv[2] is_test = int(sys.argv[3]) if len(sys.argv) > 3 else 0 tid = sys.argv[4] if len(sys.argv) > 4 else "" # 从XML文件名提取申报类型（鲁棒解析） xml_basename = os.path.splitext(os.path.basename(xml_path))[0] declare_type = parse_declare_type(xml_basename) else: # 原始调用方式：python run_ericdemo_year.py type_param = sys.argv[1] year_param = sys.argv[2] is_test = int(sys.argv[3]) tid = sys.argv[4] if len(sys.argv) > 4 else "" # 构建XML文件路径 xml_path = os.path.join(work_dir, f'{type_param}_{year_param}.xml') # 构建数据类型版本（ZMDO 不需要年份） if type_param.upper() == 'ZMDO': declare_type = 'ZMDO' else: declare_type = f"{type_param}_{year_param}" # 检查XML文件是否存在 if not os.path.exists(xml_path): error_msg = f"XML文件不存在: {xml_path}" logger.error(error_msg) print(json.dumps({"success": False, "pdf_path": "", "return_info": error_msg})) sys.exit(1) except Exception as e: error_msg = f"参数处理出错: {str(e)}" logger.error(error_msg) print(json.dumps({"success": False, "pdf_path": "", "return_info": error_msg})) sys.exit(1) # 执行申报 logger.info(f"开始处理申报: 类型={declare_type}, XML路径={xml_path}, 测试模式={is_test}, TID={tid}") result = eric_run(xml_path, declare_type, is_test, tid) # 输出结果 print(json.dumps(result)) sys.exit(0 if result["success"] else 1)