import sysimport loggingfrom logginghandlers import TimedRotatingFileHandlerimport osimport ospathimport shutilimport socketimport timeimport errnoimport tarfilefrom config import app_name app_root_di
class AppLogger: """ Application Logger class to manage logging """
def __init__(self):
"""
Initialize the logger with default settings
"""
self.logger = logging.getLogger(app_name)
self.logger.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# Log to console
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.DEBUG)
console_handler.setFormatter(formatter)
self.logger.addHandler(console_handler)
# Log to file
log_dir = os.path.join(app_root_dir, 'logs')
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_file = os.path.join(log_dir, app_name + '.log')
file_handler = TimedRotatingFileHandler(log_file, when='midnight', backupCount=7)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
def info(self, message):
"""
Log an info message
"""
self.logger.info(message)
def debug(self, message):
"""
Log a debug message
"""
self.logger.debug(message)
def error(self, message):
"""
Log an error message
"""
self.logger.error(message)
def warning(self, message):
"""
Log a warning message
"""
self.logger.warning(message)
class AppUtils: """ Application Utility class for common functions """
@staticmethod
def get_hostname():
"""
Get the hostname of the machine running the application
"""
return socket.gethostname()
@staticmethod
def get_time_str():
"""
Get the current time as a formatted string
"""
return time.strftime('%Y%m%d%H%M%S')
@staticmethod
def get_area_config(area_code):
"""
Get the configuration for a specific area
"""
return area_config.get(area_code)
@staticmethod
def get_center_config(center_code):
"""
Get the configuration for a specific center
"""
return center_config.get(center_code)
@staticmethod
def get_his_config():
"""
Get the configuration for the HIS system
"""
return his_config
@staticmethod
def get_db_conn(db_name):
"""
Get a database connection object for the specified database
"""
if db_driver == 'oracle':
conn_string = db_config.get(db_name).get('conn_string')
conn = DBManager.connect(conn_string)
return conn.cursor()
elif db_driver == 'sqlite3':
db_file = db_config.get(db_name).get('db_file')
conn = DBManager.connect(db_file)
return conn.cursor()
else:
return None
@staticmethod
def get_hdb_conn():
"""
Get a connection object for the HDB system
"""
if hdb_driver == 'mssql':
conn_string = hdb_config.get('conn_string')
conn = pyodbc.connect(conn_string)
return conn.cursor()
elif hdb_driver == 'mysql':
conn_string = hdb_config.get('conn_string')
conn = MySQLdb.connect(conn_string)
return conn.cursor()
else:
return None
@staticmethod
def generate_archive(source_dir, archive_file):
"""
Generate a tar archive of a directory
"""
with tarfile.open(archive_file, "w:gz") as tar:
tar.add(source_dir, arcname=os.path.basename(source_dir))
@staticmethod
def delete_files(files):
"""
Delete a list of files
"""
for file_path in files:
try:
os.remove(file_path)
except OSError as e:
if e.errno != errno.ENOENT:
raise
@staticmethod
def copy_files(source_files, dest_dir):
"""
Copy a list of files to a destination directory
"""
for source_file in source_files:
try:
shutil.copy2(source_file, dest_dir)
except OSError as e:
if e.errno != errno.ENOENT:
rais
原文地址: https://www.cveoy.top/t/topic/cnTz 著作权归作者所有。请勿转载和采集!