db_handler.py
· 1.7 KiB · Python
原始文件
import sqlite3
# 初始化資料庫及表格
def initialize_db(db_name="database.db", table_name="entries"):
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
cursor.execute(
f"""
CREATE TABLE IF NOT EXISTS {table_name} (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
link TEXT NOT NULL UNIQUE,
published TEXT,
description TEXT
)
"""
)
conn.commit()
conn.close()
print(f"資料庫 `{db_name}` 和表格 `{table_name}` 初始化完成。")
# 插入新資料
def insert_entry(entry, db_name="database.db", table_name="entries"):
try:
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
cursor.execute(
f"""
INSERT INTO {table_name} (title, link, published, description)
VALUES (?, ?, ?, ?)
""",
(
entry["title"],
entry["link"],
entry.get("published", None),
entry.get("description", None),
),
)
conn.commit()
print(f"成功插入新條目: {entry['title']}")
except sqlite3.IntegrityError:
print(f"條目已存在，跳過插入: {entry['link']}")
finally:
conn.close()
# 測試範例
if __name__ == "__main__":
# 初始化資料庫
initialize_db(db_name="example.db", table_name="feed_entries")
# 測試插入資料
test_entry = {
"title": "範例標題",
"link": "https://example.com/item1",
"published": "2025-01-08",
"description": "這是一個範例描述。",
}
insert_entry(test_entry, db_name="example.db", table_name="feed_entries")
| 1 | import sqlite3 |
| 2 | |
| 3 | |
| 4 | # 初始化資料庫及表格 |
| 5 | def initialize_db(db_name="database.db", table_name="entries"): |
| 6 | conn = sqlite3.connect(db_name) |
| 7 | cursor = conn.cursor() |
| 8 | cursor.execute( |
| 9 | f""" |
| 10 | CREATE TABLE IF NOT EXISTS {table_name} ( |
| 11 | id INTEGER PRIMARY KEY AUTOINCREMENT, |
| 12 | title TEXT NOT NULL, |
| 13 | link TEXT NOT NULL UNIQUE, |
| 14 | published TEXT, |
| 15 | description TEXT |
| 16 | ) |
| 17 | """ |
| 18 | ) |
| 19 | conn.commit() |
| 20 | conn.close() |
| 21 | print(f"資料庫 `{db_name}` 和表格 `{table_name}` 初始化完成。") |
| 22 | |
| 23 | |
| 24 | # 插入新資料 |
| 25 | def insert_entry(entry, db_name="database.db", table_name="entries"): |
| 26 | try: |
| 27 | conn = sqlite3.connect(db_name) |
| 28 | cursor = conn.cursor() |
| 29 | cursor.execute( |
| 30 | f""" |
| 31 | INSERT INTO {table_name} (title, link, published, description) |
| 32 | VALUES (?, ?, ?, ?) |
| 33 | """, |
| 34 | ( |
| 35 | entry["title"], |
| 36 | entry["link"], |
| 37 | entry.get("published", None), |
| 38 | entry.get("description", None), |
| 39 | ), |
| 40 | ) |
| 41 | conn.commit() |
| 42 | print(f"成功插入新條目: {entry['title']}") |
| 43 | except sqlite3.IntegrityError: |
| 44 | print(f"條目已存在，跳過插入: {entry['link']}") |
| 45 | finally: |
| 46 | conn.close() |
| 47 | |
| 48 | |
| 49 | # 測試範例 |
| 50 | if __name__ == "__main__": |
| 51 | # 初始化資料庫 |
| 52 | initialize_db(db_name="example.db", table_name="feed_entries") |
| 53 | |
| 54 | # 測試插入資料 |
| 55 | test_entry = { |
| 56 | "title": "範例標題", |
| 57 | "link": "https://example.com/item1", |
| 58 | "published": "2025-01-08", |
| 59 | "description": "這是一個範例描述。", |
| 60 | } |
| 61 | insert_entry(test_entry, db_name="example.db", table_name="feed_entries") |
| 62 |