python

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from dataclasses import replace

import pandas as pd
import oracledb
import configparser
import pathlib
import os


class ConnDB():
    def __init__(self ,table_name , sql_file_name , edate , config_path='../config/DB_config.ini', section='oracle' ):
        self.config_path = config_path
        self.section = section
        self.table_name = table_name
        self.sql_file_name = sql_file_name
        self.edate = edate
        self.base_dir = "../Result_File/"
        self.date_dir = edate.replace('-', '')
        self.target_folder_path = f"{self.base_dir}{self.date_dir}"
        self.excel_filename = f"{self.base_dir}{self.date_dir}/{self.table_name}_{self.date_dir}.xlsx"
        self.chinese_table_name = self.get_chinese_table_name(table_name)
        self.chinese_table_name = f"{self.base_dir}{self.date_dir}/{self.chinese_table_name}"

    def get_chinese_table_name(self, table_name):
        """
        根據(jù)英文表名獲取對應(yīng)的中文名稱
        """
        table_name_mapping = {
            "DM_GZW_BANK_ACCOUNT": "銀行賬戶",
            "DM_GZW_BANK_RECORD": "資金結(jié)算",
            "DM_GZW_BANK_CONTRACT": "銀行貸款",
            "DM_GZW_COMPANY": "應(yīng)付債券",
            "DM_GZW_BILLPAY": "應(yīng)付票據(jù)",
            "DM_GZW_BILLREC": "應(yīng)收票據(jù)",
            "DM_GZW_WARRANT": "擔(dān)保",
            "DM_GZW_CREDIT": "信用證",
            "DM_GZW_GUARANTEE": "保函",
            "DM_GZW_PRODUCT": "金融投資業(yè)務(wù)",
            "DM_GZW_SCM": "供應(yīng)鏈金融",
            "DM_GZW_CWGS": "財(cái)務(wù)公司附",
            "DM_GZW_PPP": "ppp附表",
            "DM_GZW_ACCOUNT_PAY": "應(yīng)付賬款",
            "DM_GZW_ACCOUNT_REC": "應(yīng)收賬款",
            "DM_GZW_CONTRACT": "合同",
            "DM_GZW_KSXX": "客商信息",
            "DM_GZW_VAT": "增值稅發(fā)票",
            "DM_GZW_JRYS": "金融衍生品"
        }

        chinese_file_name =  table_name_mapping.get(table_name)
        chinese_file_name = f"057_中國中煤能源集團(tuán)有限公司_{chinese_file_name}_01_{self.date_dir}.csv"
        return  chinese_file_name

    def get_connect(self):
        """讀取數(shù)據(jù)庫配置信息"""
        parser = configparser.ConfigParser()
        config_file = pathlib.Path(self.config_path)

        if not config_file.is_file():
            raise FileNotFoundError(f"配置文件 {self.config_path} 不存在")

        parser.read(self.config_path, encoding='utf-8')

        if not parser.has_section(self.section):
            raise ValueError(f"在配置文件中未找到 section: {self.section}")

        host = parser.get(self.section, 'host', fallback=None)
        port = parser.getint(self.section, 'port', fallback=1521)
        service_name = parser.get(self.section, 'service_name', fallback=None)
        user = parser.get(self.section, 'user', fallback=None)
        password = parser.get(self.section, 'password', fallback=None)

        print(f"連接模式:Thin , 連接IP:{host} , 連接端口:{port} , 連接用戶名:{user} , 開始測試連接...")

        DSN = f"{host}:{port}/{service_name}"

        conn = oracledb.connect(
            user=user,
            password=password,
            dsn=DSN
        )

        if not conn:
            raise ConnectionError("數(shù)據(jù)庫連接失敗，請檢查配置和網(wǎng)絡(luò)連接")
        else:
            print('Oracle數(shù)據(jù)庫連接成功!')

        return conn


    def query_date(self):
        conn = self.get_connect()



        try:
            cursor_bjg = conn.cursor()
            cursor = conn.cursor()
            #column_names = [desc[0] for desc in cursor.description]

            if  self.table_name != '':

                table_sql = f"SELECT COLMN_NAME , COLMN_CHINESE FROM GZW_TABLE_TEMPLATE WHERE TABLE_NAME =  '{self.table_name}'  ORDER BY ORDER_NUM ASC  "
                print(table_sql)
                cursor_bjg.execute(table_sql)
                rows = cursor_bjg.fetchall()
                sql_aliases = [f"{row[0]} AS \"{row[1]}\"" for row in rows]
                query_sql = ", ".join(sql_aliases)
                query_sql = f"SELECT  {query_sql} FROM {self.table_name}  WHERE DAYIDX = DATE '{self.edate}'"
                print(query_sql)
                result = cursor.execute(query_sql)
                return result


            elif   self.file_name != '':
                self.excel_filename =  f"../Result_File/{self.file_name}_{self.edate}.xlsx"
                with open(self.file_name, 'r') as file:
                    sql_str = file.read()
                    sql_str = sql_str.replace('${edate}', self.edate)
                    sql_str = sql_str.replace(';', '')
                    print(sql_str)
                    cursor.execute(sql_str)

            else:
                raise '表名和文件名都不存在'

            # df.to_excel(excel_filename, index=False)
            # print(f"數(shù)據(jù)已導(dǎo)出到: {excel_filename}")



        except Exception as e:
            print(f"執(zhí)行 SQL 時(shí)出錯(cuò): {e}")
            # 如果有更詳細(xì)的錯(cuò)誤信息，可以打印出來
            if hasattr(e, 'args') and e.args:
                print(f"錯(cuò)誤詳情: {e.args}")

    def save_file(self):
        result = self.query_date()
        if not result:
            print("沒有查詢結(jié)果！")



        else:
            if not os.path.exists(self.target_folder_path):
                try:
                    os.makedirs(self.target_folder_path)  # 使用makedirs可以同時(shí)創(chuàng)建不存在的父目錄
                    print(f"已成功創(chuàng)建文件夾: {self.target_folder_path}")
                except OSError as e:
                    print(f"創(chuàng)建文件夾時(shí)出錯(cuò): {e}")
            else:
                print(f"找到文件夾: {self.target_folder_path}")


        column_names = [desc[0] for desc in result.description]
        rows = result.fetchall()
        df = pd.DataFrame(rows, columns=column_names)

        df.to_excel(self.excel_filename, index=False)
        df.to_csv(self.chinese_table_name, index=False, encoding="utf-8")

        print(f"結(jié)果已寫入 {self.excel_filename}文件 , {self.chinese_table_name}")

from OracleODBC import ConnDB

table_name = 'DM_GZW_BANK_ACCOUNT'
sql_file_name = ''
edate = '2025-08-27'


conn1 =  ConnDB(table_name = table_name , sql_file_name='',  edate = edate)

reuslt = conn1.save_file()