大學期間大部分都專攻韌體領域的技術與知識，就連大四也都是在實習韌體工程師，對於軟體領域的點點滴滴基本上可以說是一竅不通，偏偏研究所的選擇到純軟體的實驗室進行學習，為了彌補軟體領域基本常識不足的問題，我設定了一些題目來練習，希望可以藉由「做中學」快速認識這個領域!
由於大學課堂上有接觸過HTML、CSS與JS的皮毛，對於前端有非常粗淺的認識，而後端則是完全沒有接觸過，因此決定先學習後端資料處理，在網路上看過許多案例之後想先學習透過Python做資料擷取，再使用SQL將資料保存下來方便後續應用或分析，廢話不多說，馬上進入正題~
本次目標:取得全台測速照相執法設置點
資料來源:政府資料開放平台
觀察資料後找到下載網址
檢視資料的內容，由圖中可以看到總共有2189筆資料，每一筆資料都有包含「設置縣市」、「設置市區鄉鎮」、「設置地址」、「管轄警局」、「管轄分局」、「經度」、「緯度」、「拍攝方向」以及「速限」
找到資料來源同時確認資料符合需求後，接著就要思考資料如何儲存。在網路上看了一些教學後我感覺SQL Server比較好入門，因此我事先安裝了SQL Server並選擇Azure Data Studio做為開發環境。關於安裝SQL Server的流程與注意事項可以參考這篇文章
根據使用需求創建資料表以存放測速執法設置點的資料
Meerkat
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE TABLE [dbo].[Speedlimit](
[SID] [bigint] NOT NULL,
[x] [decimal](9, 6) NOT NULL,
[y] [decimal](9, 6) NOT NULL,
[direction] [nvarchar](max) NOT NULL,
[speedlimit] [int] NOT NULL,
[lastmodified] [datetime2](0) NOT NULL,
[insertDT] [datetime2](0) NOT NULL,
[status] [bit] NOT NULL,
CONSTRAINT [PK_Speedlimit] PRIMARY KEY CLUSTERED
(
[SID] ASC
)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY] TEXTIMAGE_ON [PRIMARY]
GO
ALTER TABLE [dbo].[Speedlimit] ADD CONSTRAINT [DEFAULT_Speedlimit_status] DEFAULT (0x01) FOR [status]
GO
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE procedure [dbo].[sp_InsertSpeedlimit](
@sid bigint = null,
@x decimal(9, 6) = null,
@y decimal(9, 6) = null,
@direction nvarchar(MAX) = null,
@speedlimit int = null
)
as
begin
declare @now datetime2(0) = GETDATE();
if not exists(select 1 from Speedlimit where x = @x and y = @y and direction = @direction and speedlimit = @speedlimit)
begin
insert into Speedlimit(sid, x, y, direction, speedlimit, lastmodified, insertDT)
values(@sid, @x, @y, @direction, @speedlimit, @now, @now)
end
else
begin
update Speedlimit set lastmodified = @now, status = 0x01
where sid = (select sid from Speedlimit where x = @x and y = @y and direction = @direction and speedlimit = @speedlimit);
end
end
GO
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import requests
import time
import json
import pyodbc
# 出現錯誤時要發送信件時呼叫
def send_email(subject, body):
sender_email = "chu891106@gmail.com"
receiver_email = "chu891106@gmail.com"
app_password = "gwlv osuk kdjz zwuw"
message = MIMEMultipart()
message["From"] = sender_email
message["To"] = receiver_email
message["Subject"] = subject
body_text = MIMEText(body, "plain")
message.attach(body_text)
try:
with smtplib.SMTP("smtp.gmail.com", 587) as server:
server.starttls()
server.login(sender_email, app_password)
server.sendmail(sender_email, receiver_email, message.as_string())
print("電子郵件通知發送成功。")
except Exception as e:
print(f"發送電子郵件通知時出錯：{str(e)}")
# 呼叫API
def call_api(open_api):
try:
packet = None
while packet is None:
try:
response = requests.get(open_api)
if response.status_code == 200:
data = response.json()
packet = json.loads(json.dumps(data, ensure_ascii=False))
return packet
else:
print(f"請求發生錯誤，HTTP狀態碼：{response.status_code}")
time.sleep(3)
except requests.exceptions.RequestException as e:
print(f"請求發生錯誤：{str(e)}")
time.sleep(3)
except Exception as e:
error_subject = "Python腳本中的錯誤"
error_message = f"腳本中發生錯誤：\n\n{str(e)}"
send_email(error_subject, error_message)
# 處理資料
def proc_points(data_list):
packet = []
if data_list["success"]:
data_list = data_list["result"]["records"][1:]
if data_list:
for index, point in enumerate(data_list):
packet.append(
{
"sid": datetime.now().strftime("%Y%m%d") + str(index + 1),
"y": point["Latitude"],
"x": point["Longitude"],
"direction": point["direct"],
"speedlimit": point["limit"],
}
)
return packet
# 透過procedure儲存資料
def save_to_NewsHandle(data_list, server_config_path="server.json"):
def handle_error(function_name, error):
error_subject = f"Error in {function_name} function"
error_message = (
f"An error occurred in the {function_name} function:\n\n{str(error)}"
)
send_email(error_subject, error_message)
print(f"An error occurred in {function_name} function: {str(error)}")
try:
with open(server_config_path, "r") as config_file:
server_config = json.load(config_file)
connection_string = (
f"DRIVER={{{server_config['drv']}}};"
f"SERVER={server_config['srv']};"
f"DATABASE={server_config['db']};"
f"UID={server_config['uid']};"
f"PWD={server_config['pwd']}"
)
connection = pyodbc.connect(connection_string)
cursor = connection.cursor()
query = """
exec NewsHandle.dbo.sp_InsertSpeedlimit
@sid = ?,
@x =?,
@y =?,
@direction =?,
@speedlimit =?
"""
new_order = ["sid", "x", "y", "direction", "speedlimit"]
transformed_data_list = [
tuple(data[field] for field in new_order) for data in data_list
]
cursor.executemany(query, transformed_data_list)
connection.commit()
except Exception as e:
handle_error("exec_procedure", e)
finally:
if cursor:
cursor.close()
if connection:
connection.close()
# 主程式
if __name__ == "__main__":
print(f'開始更新測速照相 -> {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
speedlimit = call_api(
"https://od.moi.gov.tw/api/v1/rest/datastore/A01010000C-000674-011"
)
speedlimit = proc_points(speedlimit)
save_to_NewsHandle(speedlimit)
print(f'測速照相更新完成 -> {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
print("-" * 50)
print()
成果
心得
此次的練習內容可能看起來沒什麼，但作為一個新上路的菜鳥在過程中可以說是相當崎嶇，光是安裝作業環境就遇到許多問題，幸好網路上有許多先進前輩的文章可以參考，特別感謝一些大神與大老不厭其煩看我敘述問題並予以解答!另外在開發過程印象最深刻的是「Python行結尾不能打；」，這一點對我來說有點苛刻，以前在用寫C++韌體時已經養成結果打「；」的習慣了，這一點需要花一些時間來調整。我自知自己能力與認知淺薄，因此決定將學習過程分享出來當作一個紀錄，若有寫得不好、做得不好的地方也歡迎各位指教，這樣我才能更快提升。
iThome鐵人賽
看影片追技術