结构化数据与非结构化数据

1# 示例记录
2structured_data = [
3    {"id": 1, "name": "Alice", "reason_visit": "General_Checkup"},
4    {"id": 2, "name": "Bob", "reason_visit": "Diabetes"}
5]
6semi_unstructured_data = [
7    {"id": 1, "notes": "生命体征未见感染迹象。皮疹由过敏引起。"},
8    {"id": 2, "review": "将药物剂量增加至...剂量。"}
9]
10
11# 预期用法
12#manager = HealthcareDataManager()
13#manager.store_structured_data(structured_data)
14#manager.store_unstructured_data(semi_unstructured_data)
15#patient_record = manager.get_patient_record(1)
16#manager.close()

1# 使用 SQLite 的示例解决方案
2import sqlite3
3from typing import List, Dict
4import json
5
6class HealthcareDataManager:
7    def __init__(self, db_name: str = "healthcare.db"):
8        """初始化数据库连接并创建表。"""
9        self.conn = sqlite3.connect(db_name)
10        self.cursor = self.conn.cursor()
11        self._create_tables()
12    
13    def _create_tables(self):
14        """如果表不存在，则创建所需表。
15        查询语句的部分内容已为你写好。"""
16        self.cursor.executescript("""
17            CREATE TABLE IF NOT EXISTS patients (
18                id INTEGER PRIMARY KEY,
19                ...
20            );
21            
22            CREATE TABLE IF NOT EXISTS patient_notes (
23                id INTEGER PRIMARY KEY,
24                patient_id INTEGER,
25                notes TEXT,
26                review TEXT,
27                FOREIGN KEY (patient_id) REFERENCES patients (id)
28            );
29        """)
30        self.conn.commit()
31    
32    def store_structured_data(self, data: List[Dict]):
33        """存储结构化患者数据。"""
34        try:
35            for record in data:
36                self.cursor.execute(
37                    #TODO - 在此处填写你的查询语句
38                )
39            self.conn.commit()
40            return True
41        except sqlite3.Error as e:
42            print(f"存储结构化数据时出错：{e}")
43            self.conn.rollback()
44            return False
45    
46    def store_unstructured_data(self, data: List[Dict]):
47        """存储非结构化患者笔记和复诊意见。"""
48        try:
49            for record in data:
50                self.cursor.execute(
51                    #TODO - 在此处填写你的查询语句
52                    (record['id'], record.get('notes'), record.get('review'))
53                )
54            self.conn.commit()
55            return True
56        except sqlite3.Error as e:
57            print(f"存储非结构化数据时出错：{e}")
58            self.conn.rollback()
59            return False
60    
61    def get_patient_record(self, patient_id: int) -> Dict:
62        """获取包含结构化和非结构化数据的完整患者记录。"""
63        try:
64           #TODO - 在此处执行你的查询
65            result = self.cursor.fetchone()
66            if result:
67                return {
68                    "id": result[0],
69                    "name": result[1],
70                    "reason_visit": result[2],
71                    "notes": result[3],
72                    "review": result[4]
73                }
74            return None
75        except sqlite3.Error as e:
76            print(f"获取患者记录时出错：{e}")
77            return None
78    
79    def close(self):
80        """关闭数据库连接。"""
81        self.conn.close()