import os import re import sys import time import shutil import logging import subprocess from datetime import datetime from logging.handlers import TimedRotatingFileHandler import bson import requests from aws_tools import AWSUpload from env import get_env from db import * from sqlmodel import select # 配置日志 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!' } # 保留原有的环境变量获取，确保兼容性 global _env _env = get_env() def eric_run(xml_path: str, declare_type: str, is_test: int): global _env 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}") logger.info(f"使用证书: {cert_file}") # 切换到eric目录执行，因为ericprint.pdf可能生成在那里 eric_dir = os.path.join(work_dir, 'eric') command = [ python_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}") # 错误结果匹配 matches = re.findall(r'Datensatz\s+nicht\s+plausibel', result_text, re.IGNORECASE) # 是否申报成功字样匹配 success_matches = re.findall(r'Verarbeitung\s+fehlerfrei|Daten wurden erfolgreich angenommen|Code>0 0 else "" if len(matches): # 如果matches的匹配项大于0, 则代表当前申报出错 logger.error(f"申报出错: {return_info}") return [flag, dest_pdf_name, return_info] elif len(success_matches): # 需要判断是否有: Serverantwort 字样 # 申报成功, 将pdf改名且复制到新的目录 # 检查当前文件夹下是否有ericprint.pdf文件 ericprint_path = os.path.join(work_dir, 'ericprint.pdf') # 即使没有PDF文件，只要服务器返回成功，也认为申报成功 xml_matches = re.findall(r"Serverantwort\s*:[\r\n]*((?:.*[\n\r]*)+)", _rtext), re.IGNORECASE return_info = xml_matches[0][0] if xml_matches and len(xml_matches[0]) > 0 else "" flag = True if os.path.isfile(ericprint_path): # 创建temp/pdf目录 pdf_dir = os.path.join(work_dir, 'temp', 'pdf') os.makedirs(pdf_dir, exist_ok=True) # 生成唯一文件名 pdf_filename = str(bson.ObjectId()) dest_pdf_name = f"{pdf_dir}/{pdf_filename}.pdf" shutil.move(ericprint_path, dest_pdf_name) logger.info(f"申报成功，PDF文件已保存至: {dest_pdf_name}") else: logger.info("申报成功，但未生成PDF文件") return [flag, dest_pdf_name, return_info] logger.warning(f"申报结果未明确: {return_info}") return [flag, dest_pdf_name, return_info] except Exception as e: # 确保切换回原工作目录 try: os.chdir(current_dir) except: pass error_msg = f"执行ERIC申报时出错: {str(e)}" logger.error(error_msg) return [False, "", error_msg] declare_mode = ['正式申报', '测试申报'] def main(): global _env with Session(engine) as session: statement = select(DeclareData).where(DeclareData.status == 0).order_by(DeclareData.id).limit(8) results = session.exec(statement) rows = results.all() if len(rows) <= 0: logger.error(f"没有申报数据") session.close() raise BaseException("没有申报数据") for row in rows: logger.info(f"==========申报信息开始==========") logger.info(f"申报ID: {row.id}") # s3 xml path xml_path = row.xml_path declare_type = row.declare_type declare_year = row.declare_year xml_name = xml_path.split('/')[-1] local_xml_path = work_dir + "/temp/xml/" + xml_name # 先判断本地文件是否存在, 不存在就下载 if not os.path.exists(local_xml_path): resp = requests.get(xml_path, stream=True) if resp.status_code == 200: with open(local_xml_path, 'wb') as f: # 以块的形式下载，避免将整个文件加载到内存 for chunk in resp.iter_content(chunk_size=4096): # 每次读取4KB f.write(chunk) if declare_type == 'UStVA' or declare_type == 'USt': _declare_type = f"{declare_type}_{declare_year}" else: _declare_type = declare_type logger.info(f"申报类型: {_declare_type}") logger.info(f"申报环境: {declare_mode[row.test]}") flag = False xml_match = "" try: [flag, dest_pdf_name, xml_match] = eric_run(local_xml_path, _declare_type, row.test) except Exception as e: logger.info(f"调用EriC申报出现错误: {str(e)}") if flag: try: # 上传PDF aws_tool = AWSUpload(_env['aws_key_id'], _env['aws_secret_key'], _env['aws_region'], _env['aws_bucket']) key_file = aws_tool.upload_file(dest_pdf_name) # 成功生成PDF logger.info("申报成功, 申报ID为: " + str(row.id)) row.status = 1 row.result_pdf = f"{_env['aws_s3_url']}{key_file}" row.result_xml = xml_match session.add(row) session.commit() # session.refresh(row) # session.close() # cur.execute("UPDATE `declaredata` SET `status`=? ,`result_pdf`=?,`result_xml`=? WHERE `id`=?", # ("1", f"{_env['aws_s3_url']}{key_file}", xml_match, row['id']) # ) except Exception as e: session.close() logger.info(f"申报成功, 但调用AWS上传文件出现错误: {str(e)}") else: logger.error("申报失败, 申报ID为: " + str(row.id)) row.status = 2 row.result_pdf = "" row.result_xml = xml_match session.add(row) session.commit() # session.refresh(row) # session.close() # cur.execute("UPDATE `declaredata` SET `status`=? ,`result_pdf`=?,`result_xml`=? WHERE `id`=?", # ("2", "", xml_match, row['id']) # ) logger.info(f"==========申报信息结束==========") time.sleep(3) if __name__ == '__main__': os_name = os.name cron_name = 'demo-cron' # Running only Linux/Unix if os_name == 'posix': with Session(engine) as session: try: statement = select(CronLock).where(CronLock.cron == cron_name).order_by(CronLock.id).limit(1) results = session.exec(statement) cron = results.first() if cron is None: _cron = CronLock() _cron.cron = cron_name _cron.lock = 0 session.add(_cron) session.commit() session.refresh(_cron) cron = _cron if int(cron.lock) == 1: logger.info("==========上一次申报未完成, 停止当前操作==========") else: logger.info("==========上一次申报已完成, 继续当前操作==========") if not os.path.exists(env_file): raise BaseException(".env文件不存在, 请先配置") load_dotenv(f'{work_dir}/.env') _env = get_env() cron.lock = 1 session.add(cron) session.commit() session.refresh(cron) try: main() finally: cron.lock = 0 session.add(cron) session.commit() session.refresh(cron) finally: logger.info("==========session.close()==========") session.close()