Spaces:
Sleeping
Sleeping
import smtplib | |
from email.mime.text import MIMEText | |
from email.mime.multipart import MIMEMultipart | |
from email.mime.image import MIMEImage | |
import streamlit as st | |
import requests | |
from bs4 import BeautifulSoup | |
import pandas as pd | |
import plotly.express as px | |
# 定義爬取數據的函數 | |
def fetch_data(hospital_url, table_id, hospital_name): | |
response = requests.get(hospital_url) | |
soup = BeautifulSoup(response.text, 'html.parser') | |
table = soup.find('table', {'id': table_id}) | |
if not table: | |
return pd.DataFrame() # 若表格未找到,返回空的DataFrame | |
rows = table.find_all('tr') | |
data = [] | |
if hospital_name == "成大醫院": | |
columns = [th.text.strip() for th in rows[0].find_all("th")] | |
columns = ['病床種類' if col == '病床類別' else col for col in columns] | |
data_rows = rows[1:] | |
for row in data_rows: | |
row_data = [td.text.strip() for td in row.find_all("td")] | |
if all(row_data): | |
data.append(row_data) | |
else: | |
columns = ['病床種類', '病床數', '住院人數', '空床數', '佔床率'] | |
for row in rows[1:]: | |
cols = row.find_all('td') | |
if len(cols) == 5: | |
row_data = [col.get_text(strip=True) for col in cols] | |
if all(row_data): | |
data.append(row_data) | |
df = pd.DataFrame(data, columns=columns) | |
if '病床數' not in df.columns and '開放床數' in df.columns: | |
df = df.rename(columns={'開放床數': '病床數'}) | |
return df | |
# Streamlit UI | |
st.title("醫院床位分配表爬取工具") | |
# 下拉式選單選擇醫院 | |
hospital_options = { | |
"台南醫院": { | |
"url": "https://www.tmh.org.tw/tmh2016/ImpBD.aspx?Kind=2", | |
"table_id": "ctl00_ContentPlaceHolder1_GV_Bed" | |
}, | |
"奇美醫院": { | |
"url": "https://www.chimei.org.tw/%E4%BD%94%E5%BA%8A%E7%8E%87%E6%9F%A5%E8%A9%A2/%E4%BD%94%E5%BA%8A%E7%8E%87%E6%9F%A5%E8%A9%A2.aspx?ihospital=10&ffloor=", | |
"table_id": "DG1" | |
}, | |
"成大醫院": { | |
"url": "https://web.hosp.ncku.edu.tw/nckm/Bedstatus/BedStatus.aspx", | |
"table_id": "GV_EmgInsure" | |
} | |
} | |
selected_hospitals = st.multiselect("選擇醫院", list(hospital_options.keys())) | |
# 手動輸入收件人的電子郵件地址 | |
to_addr_list = [] | |
email_input = st.text_input("請輸入收件人的電子郵件地址,使用逗號分隔多個地址") | |
if email_input: | |
to_addr_list = [email.strip() for email in email_input.split(",")] | |
# 當用戶按下按鈕時,開始爬取數據 | |
if st.button("爬取資料並發送郵件"): | |
st.write("正在爬取資料...") | |
all_data = pd.DataFrame() | |
for hospital_name in selected_hospitals: | |
hospital_data = hospital_options[hospital_name] | |
df = fetch_data(hospital_data["url"], hospital_data["table_id"], hospital_name) | |
if df.empty: | |
st.warning(f"{hospital_name} 的數據爬取結果為空,請檢查是否存在問題。") | |
else: | |
df['醫院'] = hospital_name | |
all_data = pd.concat([all_data, df], ignore_index=True) | |
if not all_data.empty: | |
st.write("爬取完成,合併的數據如下:") | |
st.dataframe(all_data) | |
# 繪製圖表 | |
st.write("正在繪製圖表...") | |
bed_column = '病床數' | |
all_data[bed_column] = pd.to_numeric(all_data[bed_column], errors='coerce') | |
bed_counts = all_data.groupby(['醫院', '病床種類'])[bed_column].sum().reset_index() | |
# 圓餅圖 | |
fig_pie = px.pie(bed_counts, values=bed_column, names='病床種類', title='各類型病床分佈', | |
hover_data=['醫院'], labels={bed_column: '病床數'}) | |
st.plotly_chart(fig_pie) | |
# 保存圓餅圖為文件 | |
fig_pie.write_image("pie_chart.png") | |
# 柱狀圖 | |
fig_bar = px.bar(bed_counts, x='醫院', y=bed_column, color='病床種類', title='醫院病床分佈', | |
labels={bed_column: '病床數'}, barmode='group') | |
st.plotly_chart(fig_bar) | |
# 保存柱狀圖為文件 | |
fig_bar.write_image("bar_chart.png") | |
# 格式化數據為電子郵件內容 | |
email_body = "以下為最新醫院床位分配數據:\n\n" | |
email_body += all_data.to_string(index=False) | |
# 構建郵件 | |
msg = MIMEMultipart() | |
msg['Subject'] = '會員通知' | |
msg['From'] = '[email protected]' | |
msg['To'] = ', '.join(to_addr_list) | |
# 添加文本內容 | |
msg.attach(MIMEText(email_body, 'plain', 'utf-8')) | |
# 添加圖表圖片 | |
for image_path in ['pie_chart.png', 'bar_chart.png']: | |
with open(image_path, 'rb') as img_file: | |
img = MIMEImage(img_file.read()) | |
img.add_header('Content-Disposition', f'attachment; filename="{image_path}"') | |
msg.attach(img) | |
# 發送郵件給所有輸入的收件人 | |
if to_addr_list: | |
st.write("正在發送郵件...") | |
smtp_server = 'smtp.gmail.com' | |
smtp_port = 587 | |
from_addr = '[email protected]' | |
pwd = 'hedm elps tksg' | |
mySMTP = smtplib.SMTP(smtp_server, smtp_port) | |
mySMTP.ehlo() | |
mySMTP.starttls() | |
mySMTP.login(from_addr, pwd) | |
try: | |
mySMTP.sendmail(from_addr, to_addr_list, msg.as_string()) | |
st.success("成功發送郵件") | |
except Exception as e: | |
st.error(f"發送郵件失敗: {e}") | |
mySMTP.quit() | |
else: | |
st.warning("請輸入至少一個有效的電子郵件地址。") | |
else: | |
st.error("沒有成功爬取任何數據。") | |