100天跟着CP学PostgreSQL+AI,第31天 :PostgreSQL 如何处理 DICOM 文件？

import psycopg2
# 连接数据库
conn = psycopg2.connect(database="mypacs", user="postgres", password="123456", host="127.0.0.1", port="5432")
cur = conn.cursor()
# 创建大对象
loid = cur.execute("SELECT lo_create()").fetchone()[0]
# 打开大对象进行写入
fd = cur.connection.lobject(loid, 'w')
# 读取DICOM文件并写入大对象
with open("patient1.dcm", "rb") as f:
    data = f.read()
    fd.write(data)
# 关闭大对象
fd.close()
# 存储大对象ID到影像表
patient_id = "P001"
study_time = "2025-06-01 10:00:00"
cur.execute("INSERT INTO dicom_images (patient_id, study_time, dicom_lob) VALUES (%s, %s, %s)", (patient_id, study_time, loid))
conn.commit()
# 读取大对象
cur.execute("SELECT dicom_lob FROM dicom_images WHERE patient_id = %s AND study_time = %s", (patient_id, study_time))
loid = cur.fetchone()[0]
fd = cur.connection.lobject(loid, 'r')
dicom_data = fd.read()
fd.close()
# 关闭数据库连接
cur.close()
conn.close()

-- 创建主表
CREATE TABLE dicom_images_main (
    id SERIAL PRIMARY KEY,
    patient_id VARCHAR(50),
    study_time TIMESTAMP,
    dicom_lob OID
);
-- 创建2025年分片表
CREATE TABLE dicom_images_2025 (
    CHECK (study_time >= '2025-01-01' AND study_time < '2026-01-01')
) INHERITS (dicom_images_main);
-- 创建2024年分片表
CREATE TABLE dicom_images_2024 (
    CHECK (study_time >= '2024-01-01' AND study_time < '2025-01-01')
) INHERITS (dicom_images_main);
-- 创建索引
CREATE INDEX idx_patient_id_2025 ON dicom_images_2025 (patient_id);
CREATE INDEX idx_study_time_2025 ON dicom_images_2025 (study_time);

-- 创建组合索引
CREATE INDEX idx_patient_time ON dicom_images_main (patient_id, study_time);
-- 创建覆盖索引
CREATE INDEX idx_cover_patient_time_type ON dicom_images_main (patient_id, study_time) INCLUDE (study_type);

dicom_metadata表（影像元数据）：
CREATE TABLE dicom_metadata (
    id SERIAL PRIMARY KEY,
    patient_id VARCHAR(50),
    patient_name VARCHAR(100),
    study_time TIMESTAMP,
    study_type VARCHAR(50),  -- CT/MRI等
    series_number INT,
    instance_number INT,
    dicom_lob OID
);

import psycopg2
import pydicom
def search_dicom_images(patient_id, start_time, end_time):
    conn = psycopg2.connect(database="mypacs", user="postgres", password="123456", host="127.0.0.1", port="5432")
    cur = conn.cursor()
    
    cur.execute("""
        SELECT dicom_lob, study_type 
        FROM dicom_metadata 
        WHERE patient_id = %s AND study_time BETWEEN %s AND %s
    """, (patient_id, start_time, end_time))
    
    results = []
    for row in cur.fetchall():
        loid = row[0]
        study_type = row[1]
        fd = cur.connection.lobject(loid, 'r')
        dicom_data = fd.read()
        fd.close()
        # 可以将dicom_data保存为文件或进行其他处理
        results.append({"study_type": study_type, "dicom_data": dicom_data})
    
    cur.close()
    conn.close()
    return results