File size: 4,458 Bytes
76fb273
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3154548
76fb273
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3154548
 
76fb273
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
3154548
76fb273
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
"""
Parse NCBI BLAST XML output into structured hit list.

Raw XML is always stored to R2 first; parsing happens from
the stored copy, never inline with the API request.
"""

import xml.etree.ElementTree as ET
from typing import List, Optional


def parse_blast_xml(raw_xml: str) -> dict:
    try:
        root = ET.fromstring(raw_xml)
    except ET.ParseError as e:
        return {"error": f"XML parse error: {e}", "hits": []}

    ns = {"": "http://www.ncbi.nlm.nih.gov"}
    query_len_el = root.find(".//BlastOutput_query-len")
    query_len = int(query_len_el.text) if query_len_el is not None else 0

    hits = []
    for iteration in root.findall(".//Iteration"):
        for hit_el in iteration.findall(".//Hit"):
            hit = _parse_hit(hit_el)
            if hit is not None:
                hits.append(hit)

    return {
        "query_length": query_len,
        "hits": hits,
        "count": len(hits),
    }


def _parse_hit(hit_el: ET.Element) -> Optional[dict]:
    acc = _text(hit_el, "Hit_accession")
    if not acc:
        return None
    hit_id = _text(hit_el, "Hit_id")
    def_line = _text(hit_el, "Hit_def")
    accession = acc
    description = def_line or ""
    if " " in def_line:
        parts = def_line.split(" ", 1)
        if parts[0] == acc or parts[0] == hit_id:
            description = parts[1] if len(parts) > 1 else ""

    organism = ""
    if "[" in description and "]" in description:
        organism = description.split("[")[-1].rstrip("]")
        description = description.split("[")[0].strip()

    hsps = hit_el.findall(".//Hsp")
    top_hsp = _parse_hsp(hsps[0]) if hsps else None

    return {
        "accession": accession,
        "id": hit_id,
        "description": description,
        "organism": organism,
        "length": int(_text(hit_el, "Hit_len") or 0),
        "score": top_hsp.get("score", 0) if top_hsp else 0,
        "bit_score": top_hsp.get("bit_score", 0) if top_hsp else 0,
        "evalue": top_hsp.get("evalue", 0) if top_hsp else 0,
        "evalue_raw": top_hsp.get("evalue_raw", "0") if top_hsp else "0",
        "identity": top_hsp.get("identity", 0) if top_hsp else 0,
        "identity_pct": top_hsp.get("identity_pct", 0) if top_hsp else 0,
        "positive": top_hsp.get("positive", 0) if top_hsp else 0,
        "gaps": top_hsp.get("gaps", 0) if top_hsp else 0,
        "alignment_length": top_hsp.get("alignment_length", 0) if top_hsp else 0,
        "query_from": top_hsp.get("query_from", 0) if top_hsp else 0,
        "query_to": top_hsp.get("query_to", 0) if top_hsp else 0,
        "hit_from": top_hsp.get("hit_from", 0) if top_hsp else 0,
        "hit_to": top_hsp.get("hit_to", 0) if top_hsp else 0,
        "query_alignment": top_hsp.get("query_alignment", "") if top_hsp else "",
        "hit_alignment": top_hsp.get("hit_alignment", "") if top_hsp else "",
        "midline": top_hsp.get("midline", "") if top_hsp else "",
    }


def _parse_hsp(hsp_el: ET.Element) -> dict:
    score = int(_text(hsp_el, "Hsp_score") or 0)
    bit_score = float(_text(hsp_el, "Hsp_bit-score") or 0)
    evalue_raw = _text(hsp_el, "Hsp_evalue") or "0"
    evalue = float(evalue_raw)
    identity = int(_text(hsp_el, "Hsp_identity") or 0)
    positive = int(_text(hsp_el, "Hsp_positive") or 0)
    gaps = int(_text(hsp_el, "Hsp_gaps") or 0)
    align_len = int(_text(hsp_el, "Hsp_align-len") or 0)
    query_from = int(_text(hsp_el, "Hsp_query-from") or 0)
    query_to = int(_text(hsp_el, "Hsp_query-to") or 0)
    hit_from = int(_text(hsp_el, "Hsp_hit-from") or 0)
    hit_to = int(_text(hsp_el, "Hsp_hit-to") or 0)
    qseq = _text(hsp_el, "Hsp_qseq") or ""
    hseq = _text(hsp_el, "Hsp_hseq") or ""
    mid = _text(hsp_el, "Hsp_midline") or ""

    identity_pct = round(identity / align_len * 100, 1) if align_len > 0 else 0

    return {
        "score": score,
        "bit_score": bit_score,
        "evalue": evalue,
        "evalue_raw": evalue_raw,
        "identity": identity,
        "identity_pct": identity_pct,
        "positive": positive,
        "gaps": gaps,
        "alignment_length": align_len,
        "query_from": query_from,
        "query_to": query_to,
        "hit_from": hit_from,
        "hit_to": hit_to,
        "query_alignment": qseq,
        "hit_alignment": hseq,
        "midline": mid,
    }


def _text(el: ET.Element, path: str) -> str:
    found = el.find(path)
    return found.text if found is not None and found.text else ""