Curator

Local FastHTML app to review each report’s OCR headings and select the sections to map

run_pipeline stops at status 'awaiting_curation' once a report is OCR’d. A person then curates the report in this app, in two steps:

  1. Clean headings. The editor lists every Markdown heading in the report. You correct the text or level of any heading that OCR got wrong. Saving rewrites the report’s page_*.md files and sets curation_status to 'headings_reviewed'.
  2. Select sections. You tick the headings whose sections should be mapped. A counter shows the size of the selection against a 15,000-token budget. Saving stores the headings in selected_headings and sets curation_status to 'sections_selected'.

The next run_pipeline call maps the selected sections.

A report’s curation_status goes from 'pending' to 'headings_reviewed' to 'sections_selected'. The app saves it in the report’s results JSON, with the selection.

Exported source 
esc_behaviour= Script("""
document.addEventListener('keydown', e => {
    if (e.key === 'Escape') htmx.ajax('GET', '/editor-reset', {target: '#editor', swap: 'innerHTML'});
});
""")
Exported source 
app, rt = fast_app(
    title='IOM Curator',
    hdrs=Theme.slate.headers(mode='light', radii='small') + [esc_behaviour],
    live=True
)

The app reads reports from BASE_PATH. The path is relative to the folder you start the app from. Like the base_path of run_pipeline, it holds md/, with the OCR’d pages, and results/, with the report JSON files. The reports stay on the machine that runs the app. To try the app on the test reports from nbs/, set BASE_PATH to 'files/test'.

app uses MonsterUI’s slate theme. esc_behaviour makes the Escape key close the editor.

Helper functions

source

get_reports

def get_reports(
    base_path:str='../data', # Folder holding `md/` and `results/`
    status:NoneType=None, # `curation_status` to keep; `None` or `'all'` keeps every report
)->list: # Reports, most recent `Year` first

Load every report saved in base_path

Exported source 
def get_reports(base_path=BASE_PATH, # Folder holding `md/` and `results/`
                status=None          # `curation_status` to keep; `None` or `'all'` keeps every report
               ) -> list:            # Reports, most recent `Year` first
    "Load every report saved in `base_path`"
    results_dir = Path(base_path)/'results'
    reports = [load_report(p.stem, base_path) for p in results_dir.glob('*.json')]
    if status and status != 'all':
        reports = [r for r in reports if r.curation_status == status]
    reports.sort(key=lambda r: r.ev.meta.get('Year', '0'), reverse=True)
    return reports

source

fmt_status

def fmt_status(
    status, # A `curation_status`, such as `'headings_reviewed'`
):

Display label for status, such as 'Headings Reviewed'

Exported source 
def fmt_status(status): # A `curation_status`, such as `'headings_reviewed'`
    "Display label for `status`, such as `'Headings Reviewed'`"
    return status.replace('_', ' ').title()

source

count_selected_tokens

def count_selected_tokens(
    r:iomeval.pipeline.Report, # Report to measure
    selected_hdgs, # Raw headings like `"## 1. Introduction ... page 4"`
)->int: # Tokens in the text `extract_selected` returns; 0 if nothing is selected

Count the tokens in the sections under selected_hdgs

Exported source 
def count_selected_tokens(r:Report,     # Report to measure
                          selected_hdgs # Raw headings like `"## 1. Introduction ... page 4"`
                         ) -> int:      # Tokens in the text `extract_selected` returns; 0 if nothing is selected
    "Count the tokens in the sections under `selected_hdgs`"
    if not selected_hdgs: return 0
    md = read_pgs(r.md_path)
    return n_tokens(extract_selected(md, selected_hdgs))

source

get_progress

def get_progress(
    reports, # Reports to count
)->tuple: # `(done, total)`, where `done` counts reports with status `'sections_selected'`

Count the curated reports in reports

Exported source 
def get_progress(reports    # Reports to count
                ) -> tuple: # `(done, total)`, where `done` counts reports with status `'sections_selected'`
    "Count the curated reports in `reports`"
    done = sum(1 for r in reports if r.curation_status == 'sections_selected')
    return done, len(reports)

source

get_headings

def get_headings(
    md_path, # Folder of a report's OCR'd `page_*.md` files
)->list: # Heading lines, `#` prefix and page suffix included, in page order

List every Markdown heading in the report at md_path

Exported source 
def get_headings(md_path    # Folder of a report's OCR'd `page_*.md` files
                ) -> list:  # Heading lines, `#` prefix and page suffix included, in page order
    "List every Markdown heading in the report at `md_path`"
    md = read_pgs(md_path)
    return re.findall(r'^#+\s+.*$', md, re.MULTILINE)

get_reports loads the three test reports. One of them still waits for curation:

reports = get_reports('files/test')
L(reports).attrgot('curation_status')
['pending', 'sections_selected', 'sections_selected']

get_progress counts the other two as done:

get_progress(reports)
(2, 3)

get_headings lists the headings that the editor shows for a report:

r = load_report('49d2fba781b6a7c0d94577479636ee6f', 'files/test')
get_headings(r.md_path)[:4]
['# Final Evaluation of the EU-IOM Joint Initiative for migrant protection and reintegration in the horn of Africa ... page 1',
 '## CONTENTS ... page 3',
 '## 1. Introduction ... page 4',
 '## 2. Background of the JI-HoA ... page 5']

count_selected_tokens measures its saved selection against the budget:

count_selected_tokens(r, r.selected_headings)
9865

Components

The page has two panels. The left panel lists the reports as ReportCards, under a ProgressBar and a StatusFilter. The right panel is the editor. It shows HeadingsEditor for a 'pending' report and SectionsSelector for any other. The components carry htmx attributes that call the routes below.

The token budget in SectionsSelector is a guide. Saving a selection works whatever its size.

source

ReportCard

def ReportCard(
    r:iomeval.pipeline.Report, # Report to show
    selected:bool=False, # Highlight the card as the report open in the editor
    status:str='all', # Current status filter, passed to the editor route to keep the list filtered
):

Card with a report’s title, status, year, short ID, PDF link and Curate button

Exported source 
status_colors = dict(pending='amber', headings_reviewed='blue', sections_selected='emerald')
Exported source 
def ReportCard(r:Report,            # Report to show
               selected:bool=False, # Highlight the card as the report open in the editor
               status:str='all'     # Current status filter, passed to the editor route to keep the list filtered
              ):
    "Card with a report's title, status, year, short ID, PDF link and Curate button"
    title = r.ev.meta.get('Title', 'Untitled')
    title_short = title[:40] + '...' if len(title) > 40 else title
    year = r.ev.meta.get('Year', 'n/a')
    status_cls = {
        'pending': 'bg-red-500 text-white',
        'headings_reviewed': 'bg-blue-500 text-white', 
        'sections_selected': 'bg-emerald-500 text-white'
    }
    highlight = 'ring-1 ring-primary' if selected else ''
    return Card(
        DivFullySpaced(
            Div(
                P(title_short, title=title, cls=TextPresets.bold_sm + ' uppercase truncate'),
                DivLAligned(
                    Span(fmt_status(r.curation_status), cls=f'uk-label {status_cls[r.curation_status]} ' + TextT.xs),
                    Small(f"{year}", cls=TextPresets.muted_sm),
                    Small(f'{r.id[:4]}...', cls=TextPresets.muted_sm),
                    A("PDF", href=r.pdf_url, target='_blank', cls=TextT.sm + ' ' + AT.primary),
                    cls='gap-2'
                ),
                cls='space-y-0.5'
            ),
            Button(
                "Curate", 
                hx_get=f'/report/{r.id}?status={status}', 
                hx_target='#editor', 
                hx_swap='innerHTML show:none',
                cls='h-6 px-2 ' + TextT.xs + ButtonT.primary)
        ),
        cls=f'mb-2 {highlight}', body_cls='p-3'
    )

source

ReportList

def ReportList(
    reports, # Reports to list
    selected_id:NoneType=None, # ID of the report open in the editor
    status:str='all', # Current status filter
    oob:bool=False, # Return it for an out-of-band swap, alongside the editor
):

Scrollable list of ReportCards, with element id report-list

Exported source 
def ReportList(reports,          # Reports to list
               selected_id=None, # ID of the report open in the editor
               status='all',     # Current status filter
               oob=False         # Return it for an out-of-band swap, alongside the editor
              ):
    "Scrollable list of `ReportCard`s, with element id `report-list`"
    return Div(
        *[ReportCard(r, selected=(r.id == selected_id), status=status) for r in reports],
        id='report-list',
        cls='overflow-y-auto max-h-[80vh] p-3',
        #hx_swap_oob='true' if oob else None
        hx_swap_oob='morph' if oob else None
        #hx_swap_oob='true show:none' if oob else None
    )

source

StatusSteps

def StatusSteps(
    status, # The report's `curation_status`
):

Steps bar with ‘Clean Headings’ and ‘Select Sections’, marking the steps already done

Exported source 
def StatusSteps(status): # The report's `curation_status`
    "Steps bar with 'Clean Headings' and 'Select Sections', marking the steps already done"
    steps = ['pending', 'headings_reviewed', 'sections_selected']
    labels = ['Clean Headings', 'Select Sections']
    current_idx = steps.index(status)
    
    return Steps(*[
        LiStep(label, cls=StepT.success if i < current_idx else StepT.neutral)
        for i, label in enumerate(labels)
    ], cls='mb-2 w-full')

source

HeadingsEditor

def HeadingsEditor(
    r:iomeval.pipeline.Report, # Report to edit
):

Form with one text input per heading of r, posted to /report/{id}/save-headings

Exported source 
def HeadingsEditor(r:Report): # Report to edit
    "Form with one text input per heading of `r`, posted to `/report/{id}/save-headings`"
    headings = get_headings(r.md_path)
    title = r.ev.meta.get('Title', 'Untitled')
    return Div(
        H5(title, cls='font-bold mb-2 text-sm'),
        StatusSteps(r.curation_status),
        Form(
            Div(*[
                Input(value=h, name=f'heading_{i}', cls='mb-1 w-full font-mono text-xs')
                for i, h in enumerate(headings)
            ], 
            cls='space-y-1 max-h-[60vh] overflow-y-auto px-2 py-2 bg-slate-50 rounded border border-slate-200'
            ),
            Hidden(name='report_id', value=r.id),
            DivCentered(Button("Save Headings", type='submit', cls=ButtonT.primary), cls='w-full'),
            hx_post=f'/report/{r.id}/save-headings',
            hx_target='#editor'
        )
    )

source

token_display

def token_display(
    tokens, # Tokens in the selected sections
    budget:int=15000, # Token budget for the selection
):

Token count, green below 80% of budget, amber below budget, and red from budget up

Exported source 
def token_display(tokens,      # Tokens in the selected sections
                  budget=15000 # Token budget for the selection
                 ):
    "Token count, green below 80% of `budget`, amber below `budget`, and red from `budget` up"
    color = 'green' if tokens < budget * 0.8 else 'amber' if tokens < budget else 'red'
    return Span(
        Strong(f"{tokens:,}"), " tokens",
        id='token-count', 
        cls=f'ml-4 text-sm text-{color}-600 w-32 inline-block'
    )

source

SectionsSelector

def SectionsSelector(
    r:iomeval.pipeline.Report, # Report whose sections to select
):

Form with a checkbox per heading of r, ticked for selected_headings, and a live token count

Exported source 
def SectionsSelector(r:Report): # Report whose sections to select
    "Form with a checkbox per heading of `r`, ticked for `selected_headings`, and a live token count"
    headings = get_headings(r.md_path)
    title = r.ev.meta.get('Title', 'Untitled')
    return Div(
        H5(title, cls='font-bold mb-2 text-sm'),
        StatusSteps(r.curation_status),
        Form(
            Div(
                Table(
                    Tbody(*[
                        Tr(
                            Td(
                                CheckboxX(name=f'hdg_{i}', checked=(h in r.selected_headings)), 
                                cls='w-8 !py-1'),
                            Td(h, cls='font-mono text-xs !py-1'),
                            cls='border-b'
                        )
                        for i, h in enumerate(headings)
                    ]),
                    cls='w-full uk-table-middle'
                ),
                cls='max-h-[60vh] overflow-y-auto bg-slate-50 rounded border border-slate-200'
            ),
            DivCentered(
                DivLAligned(
                    Button(
                        "Save Selection", 
                        type='submit', 
                        cls=ButtonT.primary,
                        hx_swap='innerHTML',
                        hx_post=f'/report/{r.id}/save-sections',
                        hx_target='#editor'
                        ),
                        token_display(count_selected_tokens(r, r.selected_headings))
                ),
                cls='mt-2'
            ),
            hx_post=f'/report/{r.id}/token-count',
            hx_trigger='change from:input[type=checkbox]',
            hx_target='#token-count',
            hx_swap='outerHTML',
        ),
        Button("← Back to Headings", hx_post=f'/report/{r.id}/reset-to-pending', hx_target='#editor', hx_swap='innerHTML', cls='mt-2')
    )

source

ProgressBar

def ProgressBar(
    reports, # Reports to count
    oob:bool=False, # Return it for an out-of-band swap
):

Progress bar of the reports with status 'sections_selected'

Exported source 
def ProgressBar(reports,  # Reports to count
                oob=False # Return it for an out-of-band swap
               ):
    "Progress bar of the reports with status `'sections_selected'`"
    done, total = get_progress(reports)
    pct = int(done / total * 100) if total else 0
    return Div(
        DivLAligned(
            Progress(value=pct, max=100, cls='mb-0'),
            P(f"{done}/{total} complete", cls=TextT.sm + 'w-48'),
            cls='gap-2'
        ),
        id='progress-bar',
        hx_swap_oob='true' if oob else None
    )

source

StatusFilter

def StatusFilter(
    current:str='all', # Selected tab: `'all'` or a `curation_status`
):

Tabs that filter the report list by curation_status

Exported source 
def StatusFilter(current='all'): # Selected tab: `'all'` or a `curation_status`
    "Tabs that filter the report list by `curation_status`"
    statuses = [('all', 'All'), ('pending', 'Pending'), ('headings_reviewed', 'Reviewed'), ('sections_selected', 'Selected')]
    return TabContainer(*[
        Li(A(label, hx_get=f'/filter?status={key}', hx_target='#report-list', hx_swap='outerHTML'),
           cls='uk-active' if key == current else '')
        for key, label in statuses
    ], alt=True, cls=TextT.sm)

Routes

Each route returns HTML fragments that htmx swaps into the page. A route that changes a report also returns the report list for an out-of-band swap. The list then shows the new status.

Saving headings rewrites the report’s page_*.md files in place with apply_hdg_fixes. It keeps no copy of the original OCR output. An edit applies to every heading with the same original text. The save sets the status to 'headings_reviewed' even when no heading changed.

source

get

def get():

Page with the report list and an empty editor

Exported source 
@rt('/')
def get():
    "Page with the report list and an empty editor"
    reports = get_reports()
    return Container(
        H2("IOMEVAL | Reports Curator", cls="text-2xl font-bold mb-6"),
        Grid(
            Card(
                DivFullySpaced(
                    H4("Reports"), 
                    Div(ProgressBar(reports), cls='w-4/5'),
                    cls='items-center'
                ),
                DivCentered(Div(StatusFilter(), id='status-filter'), cls='w-2/3 mx-auto'),
                #Div(*[ReportCard(r) for r in reports], id='report-list', cls='overflow-y-auto max-h-[80vh]')),
                ReportList(reports)),
            Card(
                H4("Editor"), 
                Div(id='editor', cls='p-2')(
                    DivCentered(P('Click "Curate" to select a report', cls="font-normal"), cls='h-40')
                ), 
                cls="h-full"
            ),
            cols=2, gap=4
        ),
        cls="p-4"
    )

source

get

def get(
    id:str, status:str='all'
):

Open a report in HeadingsEditor while it is 'pending', and in SectionsSelector otherwise

Exported source 
@rt('/report/{id}')
def get(id:str, status:str='all'):
    "Open a report in `HeadingsEditor` while it is `'pending'`, and in `SectionsSelector` otherwise"
    r = load_report(id, BASE_PATH)
    reports = get_reports(status=status)
    
    editor = HeadingsEditor(r) if r.curation_status == 'pending' else SectionsSelector(r)
    
    return (
        ReportList(reports, selected_id=id, status=status, oob=True),
        editor
    )

source

post

async def post(
    id:str, req:starlette.requests.Request
):

Apply the heading edits to every page of the report and set its status to 'headings_reviewed'

Exported source 
@rt('/report/{id}/save-headings')
async def post(id:str, req:Request):
    "Apply the heading edits to every page of the report and set its status to `'headings_reviewed'`"
    form = await req.form()
    r = load_report(id, BASE_PATH)
    original_headings = get_headings(r.md_path)
    
    lut_fixes = {}
    for i, orig in enumerate(original_headings):
        edited = form.get(f'heading_{i}', orig)
        if edited != orig: lut_fixes[orig] = edited
    
    if lut_fixes:
        for pg_path in sorted(r.md_path.glob('page_*.md')):
            content = pg_path.read_text()
            fixed = apply_hdg_fixes(content, lut_fixes)
            pg_path.write_text(fixed)
    
    r.curation_status = 'headings_reviewed'
    r.save()
    
    reports = get_reports()
    return (
        ReportList(reports, selected_id=id, oob=True),
        SectionsSelector(r)
    )

source

post

def post(
    id:str
):

Send the report back to the headings step, keeping its saved selection

Exported source 
@rt('/report/{id}/reset-to-pending')
def post(id:str):
    "Send the report back to the headings step, keeping its saved selection"
    r = load_report(id, BASE_PATH)
    r.curation_status = 'pending'
    r.save()
    
    reports = get_reports()
    return (
        ReportList(reports, selected_id=id, oob=True),
        HeadingsEditor(r),
        ProgressBar(reports, oob=True)
    )

source

post

async def post(
    id:str, req:starlette.requests.Request
):

Token count of the ticked headings, recomputed on each checkbox change

Exported source 
@rt('/report/{id}/token-count')
async def post(id:str, req:Request):
    "Token count of the ticked headings, recomputed on each checkbox change"
    form = await req.form()
    r = load_report(id, BASE_PATH)
    
    selected = [int(k.split('_')[1]) for k in form.keys() if k.startswith('hdg_')]
    headings = get_headings(r.md_path)
    selected_hdgs = [headings[i] for i in selected]
    
    tokens = count_selected_tokens(r, selected_hdgs)

    return token_display(tokens)

source

post

async def post(
    id:str, req:starlette.requests.Request
):

Save the ticked headings as selected_headings and set the status to 'sections_selected'

Exported source 
@rt('/report/{id}/save-sections')
async def post(id:str, req:Request):
    "Save the ticked headings as `selected_headings` and set the status to `'sections_selected'`"
    form = await req.form()
    r = load_report(id, BASE_PATH)
    headings = get_headings(r.md_path)
    
    selected = [int(k.split('_')[1]) for k in form.keys() if k.startswith('hdg_')]
    r.selected_headings = [headings[i] for i in selected]
    r.curation_status = 'sections_selected'
    r.save()
    
    reports = get_reports()

    return (
        ReportList(reports, selected_id=id, oob=True),
        DivCentered(P('Click "Curate" to select a report', cls="font-normal"), cls='h-40'),
        ProgressBar(reports, oob=True)
    )

source

get

def get(
    status:str='all'
):

Filter the report list by status and close the editor

Exported source 
@rt('/filter')
def get(status:str='all'):
    "Filter the report list by `status` and close the editor"
    reports = get_reports(status=status)
    return (
        #Div(*[ReportCard(r, status=status) for r in reports], id='report-list'),
        ReportList(reports, status=status),
        Div(StatusFilter(current=status), id='status-filter', hx_swap_oob='true'),
        Div(
            DivCentered(P('Click "Curate" to select a report', cls="font-normal"), cls='h-40'),
            id='editor', cls='p-2', hx_swap_oob='true'
        )
    )

source

get

def get():

Close the editor. The Escape key calls this route.

Exported source 
@rt('/editor-reset')
def get():
    "Close the editor. The Escape key calls this route."
    reports = get_reports()
    return (
        ReportList(reports, oob=True),
        DivCentered(P('Click "Curate" to select a report', cls="font-normal"), cls='h-40')
    )

Server

Start the app from a folder next to your data folder, then open http://localhost:5001:

python -m iomeval.curator --port 5001

main serves app with uvicorn, which reloads the app when its code changes.

source

main

def main(
    host:str='0.0.0.0', # Network interface to listen on
    port:int=5001, # Port to listen on
):

Serve the curator app, reading reports from BASE_PATH

Exported source 
@call_parse
def main(host:str='0.0.0.0', # Network interface to listen on
         port:int=5001       # Port to listen on
        ):
    "Serve the curator app, reading reports from `BASE_PATH`"
    serve(appname="iomeval.curator", host=host, port=port)

In a notebook, JupyUvi serves the app in the background, and srv.stop() stops it:

srv = JupyUvi(app)
srv.stop()