import os import re import sys import time from logging.handlers import TimedRotatingFileHandler import bson import shutil import logging import requests import subprocess from aws_tools import AWSUpload from env import get_env from db import * from sqlmodel import select work_dir = os.getcwd() env_file = f'{work_dir}/.env' global _env # Create a logger object logger = logging.getLogger('de-declare') logger.setLevel(logging.INFO) # Set the logger's level to INFO handler = TimedRotatingFileHandler( f'{work_dir}/logs/declare.log', when='midnight', interval=1, ) formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s') handler.setFormatter(formatter) # Add the handler to the logger logger.addHandler(handler) 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 = f"{work_dir}/cert/{_env['test_cert_file']}" passwd = f"{_env['test_cert_pass']}" # else: # cert_file = f"{work_dir}/cert/{_env['prod_cert_file']}" # passwd = f"{_env['prod_cert_pass']}" result = subprocess.run([ python_executable, f"{work_dir}/eric/ericdemo.py", "-v", declare_type, "-x", xml_path, "-c", cert_file, "-p", passwd ], capture_output=True, text=True, input='\n') result_text = result.stdout dest_pdf_name = '' # 错误结果匹配 matches = re.findall(r'Datensatz\s+nicht\s+plausibel', result_text, re.IGNORECASE) # 是否申报成功字样匹配 success_matches = re.findall(r'Verarbeitung\s+fehlerfrei', result_text, re.IGNORECASE) _rtext = result_text.replace("\r\n", "") xml_matches = re.findall(r"Rückgabe\s*:[\r\n]*((?:.*[\n\r]*)+)", _rtext), re.IGNORECASE if len(matches): # 如果matches的匹配项大于0, 则代表当前申报出错, return [flag, dest_pdf_name, xml_matches[0][0]] elif len(success_matches): # 需要判断是否有: Serverantwort 字样 # 申报成功, 将pdf改名且复制到新的目录 if os.path.isfile(f"{work_dir}/ericprint.pdf"): xml_matches = re.findall(r"Serverantwort\s*:[\r\n]*((?:.*[\n\r]*)+)", _rtext), re.IGNORECASE flag = True pdf_filename = str(bson.ObjectId()) dest_pdf_name = f"{work_dir}/temp/pdf/{pdf_filename}.pdf" shutil.move(f"{work_dir}/ericprint.pdf", dest_pdf_name) return [flag, dest_pdf_name, xml_matches[0][0]] return [flag, dest_pdf_name, xml_matches[0][0]] 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()