import streamlit as st import requests from bs4 import BeautifulSoup import pandas as pd import time import base64 from io import BytesIO # 設置頁面配置 st.set_page_config( page_title="台灣證券交易所重大訊息爬蟲", page_icon="📊", layout="wide" ) # 添加標題和說明 st.title("台灣證券交易所重大訊息爬蟲") st.markdown("這個應用程式會從台灣證券交易所網站爬取上市公司的重大訊息公告。") # 添加側邊欄控制 with st.sidebar: st.header("設定") auto_refresh = st.checkbox("啟用自動刷新", value=False) refresh_interval = st.slider("刷新間隔 (分鐘)", 1, 60, 15) if auto_refresh else 0 max_results = st.slider("顯示結果數量", 5, 50, 20) st.header("篩選器") filter_enabled = st.checkbox("啟用關鍵字篩選", value=False) filter_keyword = st.text_input("關鍵字") if filter_enabled else "" st.header("關於") st.info("此應用程式從台灣證券交易所獲取重大訊息公告，僅供參考用途。") # 下載CSV功能 def get_csv_download_link(df, filename="data.csv"): """生成CSV下載鏈接""" csv = df.to_csv(index=False, encoding='utf-8-sig') b64 = base64.b64encode(csv.encode()).decode() href = f'下載 CSV 檔案' return href # 爬蟲功能 def fetch_mops_announcements(): """從台灣證券交易所爬取重大訊息""" # 設定請求URL url = "https://mopsov.twse.com.tw/mops/web/t05sr01_1" # 設定請求頭，模擬瀏覽器行為 headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", "Accept-Language": "zh-TW,zh;q=0.9,en-US;q=0.8,en;q=0.7", "Referer": "https://mopsov.twse.com.tw/mops/web/index" } with st.spinner('正在獲取資料...'): try: # 發送GET請求獲取頁面 response = requests.get(url, headers=headers, timeout=10) response.encoding = 'utf-8' # 確保正確編碼 if response.status_code != 200: st.error(f"請求失敗，狀態碼：{response.status_code}") return parse_example_data() # 解析HTML內容 soup = BeautifulSoup(response.text, 'html.parser') # 查找表格內容 table = soup.find('table', class_='hasBorder') if not table: st.warning("找不到目標表格，使用範例數據") return parse_example_data() # 查找表格體 tbody = table.find('tbody', id='tab2') if not tbody: st.warning("找不到tbody#tab2，嘗試直接查找tr元素") rows = table.find_all('tr')[1:] # 跳過表頭行 else: rows = tbody.find_all('tr') # 驗證是否找到行 if not rows: st.warning("找不到任何資料行，使用範例數據") return parse_example_data() # 準備數據列表 data = [] # 解析每行數據 for row in rows: cols = row.find_all('td') if len(cols) >= 5: company_code = cols[0].text.strip() company_name = cols[1].text.strip() announce_date = cols[2].text.strip() announce_time = cols[3].text.strip() # 處理主旨 - 可能在按鈕的title屬性中 subject_btn = cols[4].find('button') if subject_btn and 'title' in subject_btn.attrs: subject = subject_btn['title'].strip() else: # 如果沒有按鈕或title屬性，直接獲取文本 subject = cols[4].text.strip() # 添加到數據列表 data.append({ '公司代號': company_code, '公司簡稱': company_name, '發言日期': announce_date, '發言時間': announce_time, '主旨': subject }) # 如果沒有收集到數據，使用範例數據 if not data: st.warning("未收集到任何數據，使用範例數據") return parse_example_data() # 創建DataFrame df = pd.DataFrame(data) st.success(f"成功獲取 {len(df)} 筆資料") return df except Exception as e: st.error(f"爬取過程發生錯誤: {str(e)}") return parse_example_data() def parse_example_data(): """使用範例數據創建DataFrame""" # 使用範例數據 data = [ {'公司代號': '1419', '公司簡稱': '新紡', '發言日期': '114/04/01', '發言時間': '12:36:37', '主旨': '公告本公司財務主管變動'}, {'公司代號': '1419', '公司簡稱': '新紡', '發言日期': '114/04/01', '發言時間': '12:36:00', '主旨': '公告本公司發言人及代理發言人異動'}, {'公司代號': '6277', '公司簡稱': '宏正', '發言日期': '114/04/01', '發言時間': '12:03:45', '主旨': '澄清媒體報導'}, {'公司代號': '2215', '公司簡稱': '匯豐汽車', '發言日期': '114/04/01', '發言時間': '12:03:03', '主旨': '公告本公司財務暨會計主管異動'}, {'公司代號': '2215', '公司簡稱': '匯豐汽車', '發言日期': '114/04/01', '發言時間': '12:00:20', '主旨': '公告本公司新任董事長'}, {'公司代號': '6414', '公司簡稱': '樺漢', '發言日期': '114/04/01', '發言時間': '11:42:52', '主旨': '澄清工商時報有關本公司之報導'}, {'公司代號': '8916', '公司簡稱': '光隆', '發言日期': '114/04/01', '發言時間': '10:56:03', '主旨': '更正公告〔113年度股利分派情形申報作業〕(含普通股及特別股)可分配盈餘及分配後期末未分配盈餘誤植(原114/3/11董事會決議無異動)'}, {'公司代號': '6597', '公司簡稱': '立誠', '發言日期': '114/04/01', '發言時間': '10:55:35', '主旨': '澄清媒體報導'} ] return pd.DataFrame(data) # 主應用邏輯 def main(): # 添加刷新按鈕 col1, col2, col3 = st.columns([1, 1, 2]) with col1: refresh_button = st.button("刷新資料") # 添加上次更新時間顯示 if 'last_update' not in st.session_state: st.session_state.last_update = None if 'data' not in st.session_state: st.session_state.data = None # 檢查是否需要刷新數據 current_time = time.time() auto_refresh_needed = ( auto_refresh and st.session_state.last_update is not None and current_time - st.session_state.last_update > refresh_interval * 60 ) if refresh_button or auto_refresh_needed or st.session_state.data is None: # 獲取資料 df = fetch_mops_announcements() st.session_state.data = df st.session_state.last_update = current_time # 顯示最後更新時間 with col2: if st.session_state.last_update is not None: last_update_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(st.session_state.last_update)) st.info(f"最後更新: {last_update_str}") # 獲取當前資料 df = st.session_state.data # 套用篩選 if filter_enabled and filter_keyword: filtered_df = df[ df['公司代號'].str.contains(filter_keyword, case=False, na=False) | df['公司簡稱'].str.contains(filter_keyword, case=False, na=False) | df['主旨'].str.contains(filter_keyword, case=False, na=False) ] if len(filtered_df) > 0: st.success(f"找到 {len(filtered_df)} 筆符合 '{filter_keyword}' 的資料") df = filtered_df else: st.warning(f"沒有找到包含 '{filter_keyword}' 的資料") # 限制顯示數量 df_display = df.head(max_results) # 顯示數據表格 st.subheader("重大訊息公告") st.dataframe(df_display, use_container_width=True) # 顯示下載連結 st.markdown(get_csv_download_link(df, "重大訊息公告.csv"), unsafe_allow_html=True) # 顯示統計信息 st.subheader("統計資訊") col1, col2 = st.columns(2) with col1: # 日期統計 date_counts = df['發言日期'].value_counts().reset_index() date_counts.columns = ['日期', '公告數量'] st.bar_chart(date_counts.set_index('日期')) with col2: # 公司統計 company_counts = df['公司簡稱'].value_counts().head(10).reset_index() company_counts.columns = ['公司', '公告數量'] st.bar_chart(company_counts.set_index('公司')) # 主旨關鍵詞統計 st.subheader("主旨關鍵詞分析") keywords = ['董事長', '財務', '主管', '澄清', '媒體', '股利', '董事會', '異動', '收購'] keyword_counts = [] for keyword in keywords: count = df['主旨'].str.contains(keyword).sum() keyword_counts.append({'關鍵詞': keyword, '出現次數': count}) keyword_df = pd.DataFrame(keyword_counts) st.bar_chart(keyword_df.set_index('關鍵詞')) # 執行主應用 if __name__ == "__main__": main()