[Python] How can I extract symptoms/diseases from a running transcription?

I'm working on a project where I'm attempting to extract medical symptoms from a running transcription. I'm using SocketIO to get mic audio and then using Whisper to transcribe the audio into text which is saved as a file or on a MongoDB database.

I came across MedSpaCy as a possible solution to my problem, but the issue is that it requires you to define the disease and symptoms as TargetRules, and it only works on completed text.

So far, I've done the basic stuff.

import medspacy
from medspacy.ner import TargetMatcher, TargetRule
nlp = medspacy.load()
target_matcher = nlp.get_pipe("medspacy_target_matcher")


The basic import is fine. However I don't know how I should implement the diseases themselves. The best possible idea I have is to keep all the defined diseases as TargetRules in a seperate file from which I import the list to use in NLP operations upon the text, which is transcribed by this function:

def transcribe_and_emit(sid: str, audio: np.ndarray):
if audio.size == 0:
return
audio = normalize_audio(audio)
try:
result = model.transcribe(
audio,
fp16=use_fp16,
language="en",
temperature=TEMPERATURE,
condition_on_previous_text=False,
beam_size=BEAM_SIZE,
verbose=False,
)
text = (result.get("text") or "").strip()
if text:
socketio.emit("transcription", {"text": text}, to=sid)

client_info = clients.get(sid)
if client_info and client_info.get("user_id"):
save_transcript(client_info["user_id"], text, source="audio")
else:
logger.info(f"Audio chunk processed but user not registered yet (sid={sid})")

except Exception as e:
logger.warning(f"Whisper error for {sid}: {e}")



def transcribe_stream(sid: str):
q = clients[sid]['q']
running = lambda: clients.get(sid, {}).get('running', False)
buf = np.array([], dtype=np.float32)

logger.info(f"Transcription thread started for {sid}")

while running():
try:
chunk, sr = q.get(timeout=1)
except queue.Empty:
continue

# Resample to 16k if needed
if sr != SAMPLERATE:
chunk = resample_if_needed(chunk, sr, SAMPLERATE)

# Accumulate
buf = np.concatenate([buf, chunk])
sec = len(buf) / SAMPLERATE

# Force decode every MIN_UTTERANCE_SEC, or if too long
if sec >= MIN_UTTERANCE_SEC or sec >= MAX_UTTERANCE_SEC:
transcribe_and_emit(sid, buf)
buf = np.array([], dtype=np.float32)

logger.info(f"Transcription thread exiting for {sid}")


and the transcript saving is done by

def save_transcript_local(user_id, text):
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{user_id}_{timestamp}.txt"
filepath = os.path.join(TRANSCRIPTS_DIR, filename)
with open(filepath, "w", encoding="utf-8") as f:
f.write(text)
return filepath

def save_transcript_mongo(user_id, text, source="audio"):
doc = {
"user_id": user_id,
"text": text,
"source": source,
"timestamp": datetime.utcnow()
}
result = collection.insert_one(doc)
return str(result.inserted_id)

def save_transcript(user_id, text, source="audio"):
if TRANSCRIPT_STORAGE in ["local", "both"]:
save_transcript_local(user_id, text)
if TRANSCRIPT_STORAGE in ["mongo", "both"]:
save_transcript_mongo(user_id, text, source)


I don't like that I'll have to code TargetRules for every single disease. Is there any way I could automate it, given a database of diseases with listed symptoms? For now let's assume it's of the form

{
_id: str
disease: str
symptoms: list
}

Continue reading...