///|
pub enum SectionKind {
Demographics
Contact
Diagnosis
Medication
Laboratory
Narrative
Unknown
} derive(Debug, Eq)
///|
pub(all) struct DocumentSection {
title : String
kind : SectionKind
start : Int
end : Int
text : String
} derive(Debug, Eq)
///|
pub(all) struct SectionSummary {
title : String
kind : SectionKind
finding_count : Int
risk : RiskLevel
} derive(Debug, Eq)
///|
pub fn section_kind_name(kind : SectionKind) -> String {
match kind {
Demographics => "demographics"
Contact => "contact"
Diagnosis => "diagnosis"
Medication => "medication"
Laboratory => "laboratory"
Narrative => "narrative"
Unknown => "unknown"
}
}
///|
pub fn classify_section(title : String) -> SectionKind {
let normalized = title.trim().to_owned().to_lower()
if normalized.contains("人口") ||
normalized.contains("demographic") ||
normalized.contains("基本信息") {
Demographics
} else if normalized.contains("联系") || normalized.contains("contact") {
Contact
} else if normalized.contains("诊断") || normalized.contains("diagnosis") {
Diagnosis
} else if normalized.contains("用药") ||
normalized.contains("medication") ||
normalized.contains("药物") {
Medication
} else if normalized.contains("检验") ||
normalized.contains("实验室") ||
normalized.contains("laboratory") ||
normalized.contains("lab") {
Laboratory
} else if normalized.contains("病史") ||
normalized.contains("history") ||
normalized.contains("note") {
Narrative
} else {
Unknown
}
}
///|
fn is_section_header(line : String) -> Bool {
let trimmed = line.trim().to_owned()
trimmed.has_suffix(":") ||
trimmed.has_suffix(":") ||
trimmed.has_suffix("#") ||
(trimmed.to_upper() == trimmed && trimmed.length() >= 3)
}
///|
pub fn sectionize(text : String) -> Array[DocumentSection] {
let lines = split_lines_with_offsets(text)
let sections = []
let mut current_title = ""
let mut current_kind = Unknown
let mut current_start = 0
let mut current_end = 0
for line in lines {
if is_section_header(line.text) {
if current_end > current_start {
sections.push({
title: current_title,
kind: current_kind,
start: current_start,
end: current_end,
text: text[current_start:current_end].to_owned(),
})
}
current_title = line.text.trim().to_owned()
current_kind = classify_section(current_title)
current_start = line.start
current_end = line.end
} else {
current_end = line.end
}
}
if current_end > current_start {
sections.push({
title: current_title,
kind: current_kind,
start: current_start,
end: current_end,
text: text[current_start:current_end].to_owned(),
})
}
sections
}
///|
pub fn sections_of_kind(
sections : Array[DocumentSection],
kind : SectionKind,
) -> Array[DocumentSection] {
sections.filter(fn(item) { item.kind == kind })
}
///|
pub fn section_summaries(
_text : String,
sections : Array[DocumentSection],
) -> Array[SectionSummary] raise DeidError {
sections.map(section => {
let findings = scan(section.text)
{
title: section.title,
kind: section.kind,
finding_count: findings.length(),
risk: highest_risk(findings),
}
})
}
///|
pub fn redact_sections(
text : String,
config : RedactionConfig,
) -> String raise DeidError {
let sections = sectionize(text)
let findings = []
for section in sections {
let result = redact_with_config(section.text, config)
findings.append(shift_findings(result.findings, section.start))
}
let (output, _) = apply_findings(text, resolve_overlaps(findings))
output
}
///|
fn literal_matches(text : String, needle : String) -> Array[Span] {
let result = []
if needle.is_empty() {
result
} else {
let mut cursor = 0
while cursor + needle.length() <= text.length() {
if text[cursor:cursor + needle.length()] == needle {
result.push({ start: cursor, end: cursor + needle.length() })
cursor += needle.length()
} else {
cursor += 1
}
}
result
}
}
///|
pub fn literal_occurrences(
text : String,
needles : Array[String],
) -> Array[Span] {
let spans = []
for needle in needles {
spans.append(literal_matches(text, needle))
}
spans.sort_by(fn(left, right) { left.start - right.start })
spans
}
///|
pub fn literal_findings(
text : String,
needles : Array[String],
kind : PhiKind,
label : String,
rule_id : String,
confidence : Int,
) -> Array[Finding] {
literal_occurrences(text, needles).map(fn(span) {
{
id: "literal-\{span.start}",
kind,
label,
start: span.start,
end: span.end,
text: text[span.start:span.end].to_owned(),
replacement: "[\{phi_kind_name(kind)}]",
rule_id,
confidence,
}
})
}
///|
pub fn keyword_findings(text : String) -> Array[Finding] {
let keywords = [
"患者", "病人", "姓名", "住址", "身份证", "电话", "Patient", "MRN",
"DOB",
]
literal_findings(
text,
keywords,
Custom("sensitive_keyword"),
"Sensitive field label",
"keyword",
40,
)
}
///|
pub fn labeled_value_findings(text : String) -> Array[Finding] raise DeidError {
let findings = []
for field in parse_fields(text) {
let candidates = scan(field.value)
findings.append(shift_findings(candidates, field.start))
}
resolve_overlaps(findings)
}
///|
pub fn line_findings(text : String) -> Map[Int, Array[Finding]] raise DeidError {
let result : Map[Int, Array[Finding]] = Map([])
for line in split_lines_with_offsets(text) {
let findings = scan(line.text)
if !findings.is_empty() {
result[line.start] = shift_findings(findings, line.start)
}
}
result
}
///|
pub fn sensitive_line_count(text : String) -> Int raise DeidError {
line_findings(text).length()
}
///|
pub fn section_summary_markdown(summaries : Array[SectionSummary]) -> String {
let lines = [
"| Section | Kind | Findings | Risk |", "| --- | --- | ---: | --- |",
]
for summary in summaries {
lines.push(
"| \{summary.title} | \{section_kind_name(summary.kind)} | \{summary.finding_count} | \{summary.risk} |",
)
}
lines.join("\n")
}
///|
pub fn section_findings(
_text : String,
section : DocumentSection,
) -> Array[Finding] raise DeidError {
shift_findings(scan(section.text), section.start)
}
///|
pub fn section_risk(
text : String,
section : DocumentSection,
) -> RiskLevel raise DeidError {
highest_risk(section_findings(text, section))
}
///|
pub fn has_sensitive_keyword(text : String) -> Bool {
!keyword_findings(text).is_empty()
}