mychartToMarkdown.py

#!/usr/bin/env python3
"""
MyChart C-CDA XML to Markdown Converter
========================================
Converts MyChart health record downloads (IHE_XDM folder structure)
into clean, LLM-friendly Markdown files for use in Claude Projects.

Usage:
    python mychart_to_markdown.py <path_to_unzipped_mychart_folder> [output_folder]

Examples:
    python mychart_to_markdown.py ~/Downloads/MyChartExport
    python mychart_to_markdown.py ~/Downloads/MyChartExport ~/Desktop/health_records

The script will:
    - Find all DOC*.XML files in IHE_XDM subfolders
    - Parse each C-CDA document
    - Extract clinical sections (medications, problems, allergies, etc.)
    - Write one Markdown file per person/subfolder
    - Also write a combined summary Markdown if multiple people are found
"""

import os
import sys
import re
import glob
from pathlib import Path
from xml.etree import ElementTree as ET
from datetime import datetime

# C-CDA XML namespaces
NS = {
    'hl7': 'urn:hl7-org:v3',
    'xsi': 'http://www.w3.org/2001/XMLSchema-instance',
}

# LOINC section codes -> friendly names (most common C-CDA sections)
SECTION_CODES = {
    '48765-2': 'Allergies & Intolerances',
    '10160-0': 'Medications',
    '11450-4': 'Problems / Active Conditions',
    '47519-4': 'Procedures',
    '30954-2': 'Lab Results',
    '8716-3':  'Vital Signs',
    '29762-2': 'Social History',
    '10157-6': 'Family History',
    '11369-6': 'Immunizations',
    '42348-3': 'Advance Directives',
    '46240-8': 'Encounters / Visits',
    '18776-5': 'Plan of Care',
    '51847-2': 'Assessment & Plan',
    '29545-1': 'Physical Exam',
    '10164-2': 'History of Present Illness',
    '11348-0': 'Past Medical History',
    '42344-2': 'Discharge Diagnosis',
    '8648-8':  'Hospital Course',
    '11535-2': 'Discharge Medications',
    '11537-8': 'Discharge Diet',
    '59772-4': 'Planned Procedure',
    '69730-0': 'Instructions',
    '61146-7': 'Goals',
    '75310-3': 'Health Concerns',
}


def find_xml_files(base_path: Path) -> dict[str, list[Path]]:
    """Find all DOC*.XML files, grouped by person subfolder."""
    persons = {}
    xdm_path = base_path / 'IHE_XDM'

    if not xdm_path.exists():
        # Maybe the user pointed directly at IHE_XDM or a subfolder
        xdm_path = base_path

    for xml_file in sorted(xdm_path.rglob('DOC*.XML')):
        person_folder = xml_file.parent.name
        persons.setdefault(person_folder, []).append(xml_file)

    # Also catch any XML files at the top level (some exports differ)
    if not persons:
        for xml_file in sorted(base_path.rglob('*.XML')):
            if xml_file.name.upper().startswith('DOC'):
                person_folder = xml_file.parent.name
                persons.setdefault(person_folder, []).append(xml_file)

    return persons


def parse_date(date_str: str) -> str:
    """Convert C-CDA date strings (e.g. 20230715, 202307151430) to readable format."""
    if not date_str:
        return ''
    date_str = date_str.strip().split('+')[0].split('-')[0]  # strip timezone
    try:
        if len(date_str) >= 8:
            dt = datetime.strptime(date_str[:8], '%Y%m%d')
            return dt.strftime('%B %d, %Y')
    except ValueError:
        pass
    return date_str


def get_text(element, path: str, ns=NS) -> str:
    """Safely get text from an XML element by path."""
    if element is None:
        return ''
    found = element.find(path, ns)
    if found is not None and found.text:
        return found.text.strip()
    return ''


def get_attr(element, path: str, attr: str, ns=NS) -> str:
    """Safely get an attribute value from a subelement."""
    if element is None:
        return ''
    found = element.find(path, ns)
    if found is not None:
        return found.get(attr, '')
    return ''


def extract_narrative_text(section_el) -> str:
    """
    Pull the human-readable narrative text block from a section.
    This is the <text> element containing HTML-like content.
    We strip tags and return clean plain text lines.
    """
    text_el = section_el.find('hl7:text', NS)
    if text_el is None:
        return ''

    # Serialize to string then strip XML/HTML tags
    raw = ET.tostring(text_el, encoding='unicode', method='xml')
    # Remove XML tags
    clean = re.sub(r'<[^>]+>', ' ', raw)
    # Collapse whitespace
    clean = re.sub(r'\s+', ' ', clean).strip()
    # Remove common boilerplate phrases
    for phrase in ['No known', 'None documented', 'None on file']:
        pass  # Keep these — they're clinically relevant
    return clean


def extract_patient_info(root) -> dict:
    """Extract patient demographics from the ClinicalDocument header."""
    info = {}

    # Patient name
    patient = root.find('.//hl7:patientRole/hl7:patient', NS)
    if patient is not None:
        name_el = patient.find('hl7:name', NS)
        if name_el is not None:
            given = ' '.join(
                el.text for el in name_el.findall('hl7:given', NS) if el.text
            )
            family = get_text(name_el, 'hl7:family')
            prefix = get_text(name_el, 'hl7:prefix')
            parts = [p for p in [prefix, given, family] if p]
            info['name'] = ' '.join(parts)

        # DOB
        dob = get_attr(patient, 'hl7:birthTime', 'value')
        if dob:
            info['dob'] = parse_date(dob)

        # Gender
        gender_code = get_attr(patient, 'hl7:administrativeGenderCode', 'displayName')
        if not gender_code:
            gender_code = get_attr(patient, 'hl7:administrativeGenderCode', 'code')
        if gender_code:
            info['gender'] = gender_code

        # Race (optional)
        race = get_attr(patient, 'hl7:raceCode', 'displayName')
        if race and race.lower() not in ('unknown', ''):
            info['race'] = race

    # Provider org
    org = root.find('.//hl7:custodian/hl7:assignedCustodian/hl7:representedCustodianOrganization/hl7:name', NS)
    if org is not None and org.text:
        info['organization'] = org.text.strip()

    # Document date
    effective_time = root.find('hl7:effectiveTime', NS)
    if effective_time is not None:
        val = effective_time.get('value', '')
        if val:
            info['document_date'] = parse_date(val)

    # Document type
    doc_type = root.find('hl7:code', NS)
    if doc_type is not None:
        info['doc_type'] = doc_type.get('displayName', 'Clinical Document')

    return info


def extract_section(section_el) -> tuple[str, str]:
    """
    Return (section_title, markdown_content) for a single section element.
    Tries structured entries first, falls back to narrative text.
    """
    # Determine section title
    code_el = section_el.find('hl7:code', NS)
    loinc = code_el.get('code', '') if code_el is not None else ''
    title_el = section_el.find('hl7:title', NS)

    if title_el is not None and title_el.text:
        section_title = title_el.text.strip()
    elif loinc in SECTION_CODES:
        section_title = SECTION_CODES[loinc]
    else:
        section_title = f'Section ({loinc})' if loinc else 'Other'

    # Get narrative text as fallback
    narrative = extract_narrative_text(section_el)

    # Try to extract structured entries
    entries = section_el.findall('hl7:entry', NS)
    lines = []

    for entry in entries:
        line = extract_entry(entry, loinc)
        if line:
            lines.append(line)

    if lines:
        content = '\n'.join(lines)
    elif narrative:
        content = narrative
    else:
        content = '_No data recorded_'

    return section_title, content


def extract_entry(entry_el, section_loinc: str) -> str:
    """Convert a single C-CDA entry element to a Markdown bullet string."""

    # --- Medications ---
    if section_loinc in ('10160-0', '11535-2'):
        sub = entry_el.find('.//hl7:manufacturedMaterial/hl7:code', NS)
        name = ''
        if sub is not None:
            name = sub.get('displayName', '')
            if not name:
                orig = sub.find('hl7:originalText', NS)
                if orig is not None and orig.text:
                    name = orig.text.strip()
        if not name:
            return ''

        # Dose
        dose_qty = entry_el.find('.//hl7:doseQuantity', NS)
        dose = ''
        if dose_qty is not None:
            val = dose_qty.get('value', '')
            unit = dose_qty.get('unit', '')
            if val:
                dose = f'{val} {unit}'.strip()

        # Route
        route_el = entry_el.find('.//hl7:routeCode', NS)
        route = route_el.get('displayName', '') if route_el is not None else ''

        # Frequency
        freq_el = entry_el.find('.//hl7:effectiveTime[@xsi:type="PIVL_TS"]//hl7:period', NS)
        freq = ''
        if freq_el is not None:
            fval = freq_el.get('value', '')
            funit = freq_el.get('unit', '')
            freq_map = {'h': 'hourly', 'd': 'daily', 'wk': 'weekly', 'mo': 'monthly'}
            if fval and funit:
                base = freq_map.get(funit, funit)
                freq = f'every {fval} {base}' if fval != '1' else f'{base}'

        # Status
        status_el = entry_el.find('.//hl7:substanceAdministration/hl7:statusCode', NS)
        status = status_el.get('code', '') if status_el is not None else ''

        parts = [f'**{name}**']
        if dose:
            parts.append(dose)
        if route:
            parts.append(f'via {route}')
        if freq:
            parts.append(f'({freq})')
        if status and status.lower() not in ('', 'active'):
            parts.append(f'[{status}]')
        return '- ' + ' — '.join(parts)

    # --- Problems / Conditions ---
    elif section_loinc in ('11450-4', '11348-0', '42344-2'):
        obs = entry_el.find('.//hl7:observation/hl7:value', NS)
        name = ''
        if obs is not None:
            name = obs.get('displayName', '')
            if not name:
                orig = obs.find('hl7:originalText', NS)
                if orig is not None and orig.text:
                    name = orig.text.strip()

        if not name:
            return ''

        # Onset date
        onset_el = entry_el.find('.//hl7:observation/hl7:effectiveTime/hl7:low', NS)
        onset = ''
        if onset_el is not None:
            onset = parse_date(onset_el.get('value', ''))

        # Status
        status_el = entry_el.find('.//hl7:observation/hl7:statusCode', NS)
        status = status_el.get('code', '') if status_el is not None else ''

        line = f'- **{name}**'
        if onset:
            line += f' (onset: {onset})'
        if status and status.lower() not in ('', 'completed', 'active'):
            line += f' [{status}]'
        return line

    # --- Allergies ---
    elif section_loinc == '48765-2':
        obs = entry_el.find('.//hl7:observation', NS)
        substance_el = entry_el.find('.//hl7:participantRole/hl7:playingEntity/hl7:code', NS)
        substance = ''
        if substance_el is not None:
            substance = substance_el.get('displayName', '')
            if not substance:
                orig = substance_el.find('hl7:originalText', NS)
                if orig is not None and orig.text:
                    substance = orig.text.strip()

        if not substance:
            # Try the act code
            act_code = entry_el.find('.//hl7:act/hl7:code', NS)
            if act_code is not None:
                substance = act_code.get('displayName', '')

        if not substance:
            return ''

        # Reaction
        reaction_el = entry_el.find('.//hl7:observation/hl7:value', NS)
        reaction = reaction_el.get('displayName', '') if reaction_el is not None else ''

        # Severity
        severity_el = entry_el.find('.//hl7:observation/hl7:interpretationCode', NS)
        severity = severity_el.get('displayName', '') if severity_el is not None else ''

        line = f'- **{substance}**'
        if reaction:
            line += f' → {reaction}'
        if severity:
            line += f' (severity: {severity})'
        return line

    # --- Immunizations ---
    elif section_loinc == '11369-6':
        code_el = entry_el.find('.//hl7:consumable/hl7:manufacturedProduct/hl7:manufacturedMaterial/hl7:code', NS)
        name = ''
        if code_el is not None:
            name = code_el.get('displayName', '')
            if not name:
                orig = code_el.find('hl7:originalText', NS)
                if orig is not None and orig.text:
                    name = orig.text.strip()
        if not name:
            return ''

        date_el = entry_el.find('.//hl7:substanceAdministration/hl7:effectiveTime', NS)
        date_str = ''
        if date_el is not None:
            date_str = parse_date(date_el.get('value', ''))

        line = f'- **{name}**'
        if date_str:
            line += f' — {date_str}'
        return line

    # --- Vital Signs ---
    elif section_loinc == '8716-3':
        obs_el = entry_el.find('.//hl7:observation', NS)
        if obs_el is None:
            # organizer -> component -> observation
            obs_el = entry_el.find('.//hl7:organizer/hl7:component/hl7:observation', NS)
        if obs_el is None:
            return ''

        name_el = obs_el.find('hl7:code', NS)
        name = name_el.get('displayName', '') if name_el is not None else ''
        value_el = obs_el.find('hl7:value', NS)
        value = ''
        if value_el is not None:
            v = value_el.get('value', '')
            u = value_el.get('unit', '')
            value = f'{v} {u}'.strip() if v else ''

        date_el = obs_el.find('hl7:effectiveTime', NS)
        date_str = ''
        if date_el is not None:
            date_str = parse_date(date_el.get('value', ''))

        if not name:
            return ''
        line = f'- **{name}**: {value}'
        if date_str:
            line += f' ({date_str})'
        return line

    # --- Procedures ---
    elif section_loinc == '47519-4':
        proc_el = entry_el.find('.//hl7:procedure/hl7:code', NS)
        if proc_el is None:
            proc_el = entry_el.find('.//hl7:act/hl7:code', NS)
        name = ''
        if proc_el is not None:
            name = proc_el.get('displayName', '')
            if not name:
                orig = proc_el.find('hl7:originalText', NS)
                if orig is not None and orig.text:
                    name = orig.text.strip()
        if not name:
            return ''

        date_el = entry_el.find('.//hl7:procedure/hl7:effectiveTime', NS)
        if date_el is None:
            date_el = entry_el.find('.//hl7:act/hl7:effectiveTime', NS)
        date_str = ''
        if date_el is not None:
            date_str = parse_date(date_el.get('value', '') or
                                  get_attr(date_el, 'hl7:low', 'value'))

        line = f'- **{name}**'
        if date_str:
            line += f' — {date_str}'
        return line

    # --- Lab Results ---
    elif section_loinc == '30954-2':
        # Labs are usually nested in organizers
        results = []
        for org in entry_el.findall('.//hl7:organizer', NS):
            panel_el = org.find('hl7:code', NS)
            panel = panel_el.get('displayName', 'Lab Panel') if panel_el is not None else 'Lab Panel'
            obs_lines = []
            for comp in org.findall('.//hl7:component/hl7:observation', NS):
                obs_name_el = comp.find('hl7:code', NS)
                obs_name = obs_name_el.get('displayName', '') if obs_name_el is not None else ''
                val_el = comp.find('hl7:value', NS)
                val = ''
                if val_el is not None:
                    v = val_el.get('value', val_el.text or '')
                    u = val_el.get('unit', '')
                    val = f'{v} {u}'.strip() if v else ''
                ref_el = comp.find('hl7:referenceRange/hl7:observationRange/hl7:text', NS)
                ref = ref_el.text.strip() if ref_el is not None and ref_el.text else ''
                interp_el = comp.find('hl7:interpretationCode', NS)
                interp = interp_el.get('displayName', interp_el.get('code', '')) if interp_el is not None else ''

                if obs_name:
                    obs_line = f'  - **{obs_name}**: {val}'
                    if ref:
                        obs_line += f' (ref: {ref})'
                    if interp and interp.upper() not in ('', 'N', 'NORMAL'):
                        obs_line += f' ⚠️ {interp}'
                    obs_lines.append(obs_line)

            if obs_lines:
                results.append(f'- **{panel}**')
                results.extend(obs_lines)

        return '\n'.join(results) if results else ''

    # --- Social History ---
    elif section_loinc == '29762-2':
        obs_el = entry_el.find('.//hl7:observation', NS)
        if obs_el is None:
            return ''
        code_el = obs_el.find('hl7:code', NS)
        name = code_el.get('displayName', '') if code_el is not None else ''
        val_el = obs_el.find('hl7:value', NS)
        value = ''
        if val_el is not None:
            value = val_el.get('displayName', val_el.text or '')
        if not name:
            return ''
        return f'- **{name}**: {value}' if value else f'- **{name}**'

    # --- Encounters ---
    elif section_loinc == '46240-8':
        enc_el = entry_el.find('.//hl7:encounter', NS)
        if enc_el is None:
            return ''
        code_el = enc_el.find('hl7:code', NS)
        name = code_el.get('displayName', 'Encounter') if code_el is not None else 'Encounter'
        date_el = enc_el.find('hl7:effectiveTime', NS)
        date_str = ''
        if date_el is not None:
            date_str = parse_date(date_el.get('value', '') or
                                  get_attr(date_el, 'hl7:low', 'value'))
        provider_el = enc_el.find('.//hl7:assignedPerson/hl7:name', NS)
        provider = ''
        if provider_el is not None:
            given = ' '.join(e.text for e in provider_el.findall('hl7:given', NS) if e.text)
            family = get_text(provider_el, 'hl7:family')
            provider = f'{given} {family}'.strip()

        line = f'- **{name}**'
        if date_str:
            line += f' — {date_str}'
        if provider:
            line += f' (with {provider})'
        return line

    # Default: fall through (narrative will be used)
    return ''


def xml_to_markdown(xml_path: Path) -> str:
    """Parse a single C-CDA XML file and return Markdown string."""
    try:
        tree = ET.parse(xml_path)
        root = tree.getroot()
    except ET.ParseError as e:
        return f'_Error parsing {xml_path.name}: {e}_\n'

    lines = []

    # Patient info block
    info = extract_patient_info(root)
    if info.get('name'):
        lines.append(f"# {info['name']}")
    else:
        lines.append(f'# Health Record: {xml_path.stem}')

    lines.append('')
    if info.get('doc_type'):
        lines.append(f"**Document Type:** {info['doc_type']}")
    if info.get('document_date'):
        lines.append(f"**Date:** {info['document_date']}")
    if info.get('organization'):
        lines.append(f"**Source:** {info['organization']}")
    if info.get('dob'):
        lines.append(f"**Date of Birth:** {info['dob']}")
    if info.get('gender'):
        lines.append(f"**Sex:** {info['gender']}")
    lines.append('')
    lines.append('---')
    lines.append('')

    # Sections
    sections = root.findall('.//hl7:component/hl7:section', NS)
    # Also catch top-level structuredBody sections
    if not sections:
        sections = root.findall(
            './/hl7:structuredBody/hl7:component/hl7:section', NS
        )

    seen_titles = set()
    for section_el in sections:
        title, content = extract_section(section_el)
        if not content or content == '_No data recorded_':
            continue
        # Deduplicate (some exports repeat sections)
        if title in seen_titles:
            continue
        seen_titles.add(title)

        lines.append(f'## {title}')
        lines.append('')
        lines.append(content)
        lines.append('')

    return '\n'.join(lines)


def process_person(person_name: str, xml_files: list[Path], output_dir: Path) -> Path:
    """Process all XML files for one person and write a combined Markdown file."""
    all_sections: dict[str, list[str]] = {}
    patient_header = ''

    for xml_file in xml_files:
        md = xml_to_markdown(xml_file)
        # Split header from sections
        parts = md.split('---\n', 1)
        if not patient_header and len(parts) == 2:
            patient_header = parts[0] + '---\n\n'
        content = parts[1] if len(parts) == 2 else md

        # Parse out sections to deduplicate/merge across files
        current_title = None
        current_lines = []
        for line in content.split('\n'):
            if line.startswith('## '):
                if current_title and current_lines:
                    all_sections.setdefault(current_title, [])
                    all_sections[current_title].extend(current_lines)
                current_title = line[3:].strip()
                current_lines = []
            else:
                current_lines.append(line)
        if current_title and current_lines:
            all_sections.setdefault(current_title, [])
            all_sections[current_title].extend(current_lines)

    # Write combined output
    output_lines = [patient_header] if patient_header else [f'# Health Record: {person_name}\n\n---\n\n']

    for title, content_lines in all_sections.items():
        # Deduplicate bullet lines within each section
        seen = set()
        deduped = []
        for line in content_lines:
            stripped = line.strip()
            if stripped and stripped not in seen:
                seen.add(stripped)
                deduped.append(line)
            elif not stripped and deduped and deduped[-1].strip():
                deduped.append(line)  # preserve spacing

        body = '\n'.join(deduped).strip()
        if body and body != '_No data recorded_':
            output_lines.append(f'## {title}\n\n{body}\n\n')

    safe_name = re.sub(r'[^\w\-_. ]', '_', person_name)
    out_path = output_dir / f'{safe_name}_health_record.md'
    out_path.write_text('\n'.join(output_lines), encoding='utf-8')
    return out_path


def main():
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)

    input_path = Path(sys.argv[1]).expanduser().resolve()
    output_dir = Path(sys.argv[2]).expanduser().resolve() if len(sys.argv) > 2 else input_path / 'markdown_output'

    if not input_path.exists():
        print(f'ERROR: Path not found: {input_path}')
        sys.exit(1)

    output_dir.mkdir(parents=True, exist_ok=True)
    print(f'\n📂 Input:  {input_path}')
    print(f'📁 Output: {output_dir}\n')

    persons = find_xml_files(input_path)

    if not persons:
        print('ERROR: No DOC*.XML files found. Make sure you point at the unzipped MyChart folder.')
        sys.exit(1)

    print(f'Found {sum(len(v) for v in persons.values())} XML file(s) for {len(persons)} person(s):\n')

    written = []
    for person_name, xml_files in sorted(persons.items()):
        print(f'  ⚕  Processing {person_name} ({len(xml_files)} file(s))...')
        out_path = process_person(person_name, xml_files, output_dir)
        written.append(out_path)
        print(f'     ✅ Written: {out_path.name}')

    print(f'\n✨ Done! {len(written)} Markdown file(s) saved to:\n   {output_dir}\n')
    print('These files are ready to upload to a Claude Project as context.\n')


if __name__ == '__main__':
    main()