主程式:Local_Stock_Main.py
Last updated
Last updated
此為程式進入點,同時控制流程走向。
所有的機敏資訊(包含資料庫連線資訊、API Token、LINE Notify Token)存放在 CSV 檔案中,程式一開始會先解密,之後將個模組所需要的資訊以參數方式傳遞。
接下來會先判斷程式執行當日在 [dbo].[Stock_WorkDay]
裡是否為股市營業日,若為營業日才接續以及。
import pyodbc
import pandas
from datetime import datetime
import csv
from module.Local_Encryption import *
from module.Local_Stock_Meta import *
from module.Local_Stock_Quote import *
from module.Local_Stock_Line_Notify import *
def main():
# 開啟 CSV 檔案
with open('C:\\Local_Stock_Main\\Local_Server.csv', 'r', encoding='utf-8') as file:
# 建立 CSV 檔案字典讀取器
reader = csv.DictReader(file)
# 讀取 CSV 檔案的內容
for row in reader:
# 將 CSV 裡面的加密資料進行解密
server_name = aes_decrypt(row['SERVER_NAME'])
database = aes_decrypt(row['DATABASE'])
account = aes_decrypt(row['ACCOUNT'])
pwd = aes_decrypt(row['PWD'])
api_token = aes_decrypt(row['API_Token'])
line_token = aes_decrypt(row['Line_Token'])
pic_folder = row['Pic_Folder']
now = datetime.today().strftime('%Y-%m-%d')
# 開啟 SQL Server 連線
cnxn = pyodbc.connect('DRIVER={SQL Server};SERVER='+server_name+';DATABASE='+database+';UID='+account+';PWD='+ pwd)
cursor = cnxn.cursor()
# 查詢當日是否為股市營業日
query = f"SELECT isWorkDay FROM [Stock_API].[dbo].[Stock_WorkDay] WHERE Date = '{now}';"
df = pandas.read_sql(query, cnxn)
for index, item in df.iterrows():
if item['isWorkDay'] == "N":
print(f"今天 {now} 非股市營業日")
# 關閉 SQL Server 連線
cursor.close()
cnxn.close()
else:
# step 2. 取得個股基本資訊
main_meta(server_name, database, account, pwd, api_token)
# step 3. 取得當日個股交易資訊
main_quote(server_name, database, account, pwd, api_token)
# step 4. 發送資料
query_table(server_name, database, account, pwd, line_token, pic_folder)
# 關閉 SQL Server 連線
cursor.close()
cnxn.close()
main()