Biomanus/biomni/tool/local_bio_cache.py

from __future__ import annotations

import csv
import json
import os
import shutil
import sqlite3
import subprocess
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


try:
    from Bio import SeqIO  # type: ignore
except Exception:  # pragma: no cover
    SeqIO = None


PROJECT_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_CACHE_ROOT = PROJECT_ROOT / "data" / "biomni_data" / "local_mcp_cache"


def _utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _cache_root() -> Path:
    raw = os.getenv("HYPOBIOOS_LOCAL_MCP_DATA")
    root = Path(raw) if raw else DEFAULT_CACHE_ROOT
    root.mkdir(parents=True, exist_ok=True)
    (root / "blast_db").mkdir(exist_ok=True)
    (root / "collections").mkdir(exist_ok=True)
    return root


def _db_path() -> Path:
    return _cache_root() / "local_bio_cache.sqlite3"


def _connect() -> sqlite3.Connection:
    conn = sqlite3.connect(_db_path())
    conn.row_factory = sqlite3.Row
    _ensure_schema(conn)
    return conn


def _ensure_schema(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS sequence_collections (
            name TEXT PRIMARY KEY,
            molecule_type TEXT NOT NULL,
            source TEXT,
            description TEXT,
            fasta_path TEXT,
            blast_db_prefix TEXT,
            created_at TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS sequences (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            collection_name TEXT NOT NULL,
            accession TEXT,
            label TEXT,
            organism TEXT,
            description TEXT,
            sequence TEXT NOT NULL,
            seq_length INTEGER NOT NULL,
            header TEXT,
            FOREIGN KEY(collection_name) REFERENCES sequence_collections(name)
        );

        CREATE INDEX IF NOT EXISTS idx_sequences_collection ON sequences(collection_name);
        CREATE INDEX IF NOT EXISTS idx_sequences_accession ON sequences(accession);
        CREATE INDEX IF NOT EXISTS idx_sequences_label ON sequences(label);

        CREATE TABLE IF NOT EXISTS tf_binding_collections (
            name TEXT PRIMARY KEY,
            source TEXT,
            description TEXT,
            table_path TEXT,
            created_at TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS tf_binding_records (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            collection_name TEXT NOT NULL,
            tf_name TEXT,
            target_gene TEXT,
            cell_type TEXT,
            biosample TEXT,
            evidence TEXT,
            score REAL,
            chrom TEXT,
            start INTEGER,
            end INTEGER,
            source TEXT,
            raw_json TEXT NOT NULL,
            FOREIGN KEY(collection_name) REFERENCES tf_binding_collections(name)
        );

        CREATE INDEX IF NOT EXISTS idx_tf_collection ON tf_binding_records(collection_name);
        CREATE INDEX IF NOT EXISTS idx_tf_name ON tf_binding_records(tf_name);
        CREATE INDEX IF NOT EXISTS idx_tf_target_gene ON tf_binding_records(target_gene);
        CREATE INDEX IF NOT EXISTS idx_tf_cell_type ON tf_binding_records(cell_type);
        """
    )
    conn.commit()


def initialize_local_bio_cache() -> dict[str, str]:
    root = _cache_root()
    with _connect():
        pass
    return {
        "cache_root": str(root),
        "sqlite_db": str(_db_path()),
    }


def list_sequence_collections() -> list[dict[str, Any]]:
    with _connect() as conn:
        rows = conn.execute(
            """
            SELECT c.name, c.molecule_type, c.source, c.description, c.fasta_path, c.blast_db_prefix, c.created_at,
                   COUNT(s.id) AS sequence_count
            FROM sequence_collections c
            LEFT JOIN sequences s ON s.collection_name = c.name
            GROUP BY c.name
            ORDER BY c.name
            """
        ).fetchall()
    return [dict(row) for row in rows]


def list_tf_binding_collections() -> list[dict[str, Any]]:
    with _connect() as conn:
        rows = conn.execute(
            """
            SELECT c.name, c.source, c.description, c.table_path, c.created_at,
                   COUNT(r.id) AS record_count
            FROM tf_binding_collections c
            LEFT JOIN tf_binding_records r ON r.collection_name = c.name
            GROUP BY c.name
            ORDER BY c.name
            """
        ).fetchall()
    return [dict(row) for row in rows]


def register_sequence_collection(
    collection_name: str,
    fasta_path: str,
    molecule_type: str = "protein",
    source: str = "local",
    description: str = "",
    rebuild: bool = True,
) -> dict[str, Any]:
    path = Path(fasta_path).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"FASTA file not found: {path}")

    records = list(_iter_fasta_records(path))
    if not records:
        raise ValueError(f"No FASTA records found in: {path}")

    collection_dir = _cache_root() / "collections" / collection_name
    collection_dir.mkdir(parents=True, exist_ok=True)
    cached_fasta = collection_dir / path.name
    if cached_fasta.resolve() != path:
        shutil.copy2(path, cached_fasta)
    else:
        cached_fasta = path

    blast_db_prefix = None
    if rebuild:
        blast_db_prefix = _maybe_build_blast_db(cached_fasta, collection_name, molecule_type)

    with _connect() as conn:
        conn.execute("DELETE FROM sequences WHERE collection_name = ?", (collection_name,))
        conn.execute(
            """
            INSERT INTO sequence_collections(name, molecule_type, source, description, fasta_path, blast_db_prefix, created_at)
            VALUES(?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(name) DO UPDATE SET
                molecule_type=excluded.molecule_type,
                source=excluded.source,
                description=excluded.description,
                fasta_path=excluded.fasta_path,
                blast_db_prefix=excluded.blast_db_prefix,
                created_at=excluded.created_at
            """,
            (
                collection_name,
                molecule_type,
                source,
                description,
                str(cached_fasta),
                blast_db_prefix,
                _utc_now(),
            ),
        )
        conn.executemany(
            """
            INSERT INTO sequences(
                collection_name, accession, label, organism, description, sequence, seq_length, header
            ) VALUES(?, ?, ?, ?, ?, ?, ?, ?)
            """,
            [
                (
                    collection_name,
                    record.accession,
                    record.label,
                    record.organism,
                    record.description,
                    record.sequence,
                    len(record.sequence),
                    record.header,
                )
                for record in records
            ],
        )
        conn.commit()

    return {
        "collection_name": collection_name,
        "sequence_count": len(records),
        "molecule_type": molecule_type,
        "source": source,
        "cached_fasta": str(cached_fasta),
        "blast_db_prefix": blast_db_prefix,
        "blast_db_built": bool(blast_db_prefix),
    }


def remove_sequence_collection(collection_name: str) -> dict[str, Any]:
    with _connect() as conn:
        row = conn.execute(
            "SELECT fasta_path, blast_db_prefix FROM sequence_collections WHERE name = ?",
            (collection_name,),
        ).fetchone()
        if row is None:
            return {"removed": False, "collection_name": collection_name, "reason": "not found"}
        conn.execute("DELETE FROM sequences WHERE collection_name = ?", (collection_name,))
        conn.execute("DELETE FROM sequence_collections WHERE name = ?", (collection_name,))
        conn.commit()

    fasta_path = row["fasta_path"]
    blast_db_prefix = row["blast_db_prefix"]
    if fasta_path:
        _best_effort_remove_path(Path(fasta_path))
    if blast_db_prefix:
        for suffix in (".pdb", ".phr", ".pin", ".pog", ".psq", ".ndb", ".nhr", ".nin", ".nog", ".nsq"):
            _best_effort_remove_path(Path(str(blast_db_prefix) + suffix))
    return {"removed": True, "collection_name": collection_name}


def local_blast_search(
    query_sequence: str,
    collection_name: str = "",
    top_k: int = 5,
    min_identity: float = 0.35,
    prefer_blast: bool = True,
) -> dict[str, Any]:
    normalized_query = _normalize_sequence(query_sequence)
    if not normalized_query:
        raise ValueError("Query sequence is empty after normalization.")

    collection = _resolve_sequence_collection(collection_name)
    if collection is None:
        raise ValueError(f"Unknown sequence collection: {collection_name or '(empty)'}")

    if prefer_blast:
        blast_result = _run_local_blast(normalized_query, collection, top_k=top_k)
        if blast_result is not None:
            blast_result["fallback_used"] = False
            return blast_result

    sequences = _load_sequences(collection["name"])
    hits = _search_sequences_python(normalized_query, sequences, top_k=top_k, min_identity=min_identity)
    return {
        "method": "python_fallback",
        "collection_name": collection["name"],
        "query_length": len(normalized_query),
        "fallback_used": True,
        "hits": hits,
    }


def local_uniprot_sequence_search(
    query_sequence: str,
    collection_name: str = "",
    top_k: int = 5,
    min_identity: float = 0.35,
) -> dict[str, Any]:
    resolved = collection_name
    if not resolved:
        for item in list_sequence_collections():
            name = str(item.get("name", "")).lower()
            source = str(item.get("source", "")).lower()
            if "uniprot" in name or "uniprot" in source:
                resolved = item["name"]
                break
    return local_blast_search(
        query_sequence=query_sequence,
        collection_name=resolved,
        top_k=top_k,
        min_identity=min_identity,
        prefer_blast=True,
    )


def register_tf_binding_table(
    collection_name: str,
    table_path: str,
    source: str = "local",
    description: str = "",
    file_format: str = "auto",
) -> dict[str, Any]:
    path = Path(table_path).expanduser().resolve()
    if not path.exists():
        raise FileNotFoundError(f"TF binding table not found: {path}")

    rows = list(_iter_table_rows(path, file_format=file_format))
    if not rows:
        raise ValueError(f"No rows found in TF binding file: {path}")

    collection_dir = _cache_root() / "collections" / collection_name
    collection_dir.mkdir(parents=True, exist_ok=True)
    cached_table = collection_dir / path.name
    if cached_table.resolve() != path:
        shutil.copy2(path, cached_table)
    else:
        cached_table = path

    normalized_rows = [_normalize_tf_row(row, source=source) for row in rows]

    with _connect() as conn:
        conn.execute("DELETE FROM tf_binding_records WHERE collection_name = ?", (collection_name,))
        conn.execute(
            """
            INSERT INTO tf_binding_collections(name, source, description, table_path, created_at)
            VALUES(?, ?, ?, ?, ?)
            ON CONFLICT(name) DO UPDATE SET
                source=excluded.source,
                description=excluded.description,
                table_path=excluded.table_path,
                created_at=excluded.created_at
            """,
            (
                collection_name,
                source,
                description,
                str(cached_table),
                _utc_now(),
            ),
        )
        conn.executemany(
            """
            INSERT INTO tf_binding_records(
                collection_name, tf_name, target_gene, cell_type, biosample, evidence, score,
                chrom, start, end, source, raw_json
            ) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            [
                (
                    collection_name,
                    row.get("tf_name"),
                    row.get("target_gene"),
                    row.get("cell_type"),
                    row.get("biosample"),
                    row.get("evidence"),
                    row.get("score"),
                    row.get("chrom"),
                    row.get("start"),
                    row.get("end"),
                    row.get("source"),
                    json.dumps(row.get("raw", {}), ensure_ascii=False),
                )
                for row in normalized_rows
            ],
        )
        conn.commit()

    return {
        "collection_name": collection_name,
        "record_count": len(normalized_rows),
        "source": source,
        "cached_table": str(cached_table),
    }


def remove_tf_binding_collection(collection_name: str) -> dict[str, Any]:
    with _connect() as conn:
        row = conn.execute(
            "SELECT table_path FROM tf_binding_collections WHERE name = ?",
            (collection_name,),
        ).fetchone()
        if row is None:
            return {"removed": False, "collection_name": collection_name, "reason": "not found"}
        conn.execute("DELETE FROM tf_binding_records WHERE collection_name = ?", (collection_name,))
        conn.execute("DELETE FROM tf_binding_collections WHERE name = ?", (collection_name,))
        conn.commit()
    if row["table_path"]:
        _best_effort_remove_path(Path(row["table_path"]))
    return {"removed": True, "collection_name": collection_name}


def query_local_tf_binding(
    tf_name: str = "",
    target_gene: str = "",
    cell_type: str = "",
    collection_name: str = "",
    top_k: int = 20,
) -> dict[str, Any]:
    clauses = []
    values: list[Any] = []
    if collection_name:
        clauses.append("collection_name = ?")
        values.append(collection_name)
    if tf_name:
        clauses.append("LOWER(COALESCE(tf_name, '')) LIKE ?")
        values.append(f"%{tf_name.lower()}%")
    if target_gene:
        clauses.append("LOWER(COALESCE(target_gene, '')) LIKE ?")
        values.append(f"%{target_gene.lower()}%")
    if cell_type:
        clauses.append("(LOWER(COALESCE(cell_type, '')) LIKE ? OR LOWER(COALESCE(biosample, '')) LIKE ?)")
        values.extend([f"%{cell_type.lower()}%", f"%{cell_type.lower()}%"])

    where_sql = f"WHERE {' AND '.join(clauses)}" if clauses else ""
    sql = (
        "SELECT collection_name, tf_name, target_gene, cell_type, biosample, evidence, score, chrom, start, end, source, raw_json "
        f"FROM tf_binding_records {where_sql} "
        "ORDER BY COALESCE(score, 0) DESC, tf_name, target_gene "
        "LIMIT ?"
    )
    values.append(top_k)
    with _connect() as conn:
        rows = conn.execute(sql, values).fetchall()

    hits = []
    for row in rows:
        payload = dict(row)
        payload["raw"] = json.loads(payload.pop("raw_json"))
        hits.append(payload)

    return {
        "collection_name": collection_name or "all",
        "filters": {"tf_name": tf_name, "target_gene": target_gene, "cell_type": cell_type},
        "hit_count": len(hits),
        "hits": hits,
    }


def local_pairwise_alignment(
    sequence_a: str,
    sequence_b: str,
    mode: str = "global",
    match_score: int = 2,
    mismatch_score: int = -1,
    gap_score: int = -2,
) -> dict[str, Any]:
    seq_a = _normalize_sequence(sequence_a)
    seq_b = _normalize_sequence(sequence_b)
    if not seq_a or not seq_b:
        raise ValueError("Both sequences must contain non-empty characters after normalization.")
    if mode not in {"global", "local"}:
        raise ValueError("mode must be 'global' or 'local'")

    alignment = _needleman_wunsch(seq_a, seq_b, match_score, mismatch_score, gap_score, local=(mode == "local"))
    return alignment


@dataclass
class SequenceRecord:
    accession: str
    label: str
    organism: str
    description: str
    sequence: str
    header: str


def _iter_fasta_records(path: Path) -> list[SequenceRecord]:
    if SeqIO is not None:
        records = []
        for entry in SeqIO.parse(str(path), "fasta"):
            header = str(entry.description)
            accession, label, organism, description = _parse_fasta_header(header)
            records.append(
                SequenceRecord(
                    accession=accession,
                    label=label,
                    organism=organism,
                    description=description,
                    sequence=_normalize_sequence(str(entry.seq)),
                    header=header,
                )
            )
        return [record for record in records if record.sequence]

    records: list[SequenceRecord] = []
    header = None
    chunks: list[str] = []
    for line in path.read_text(encoding="utf-8").splitlines():
        if line.startswith(">"):
            if header is not None:
                accession, label, organism, description = _parse_fasta_header(header)
                sequence = _normalize_sequence("".join(chunks))
                if sequence:
                    records.append(
                        SequenceRecord(
                            accession=accession,
                            label=label,
                            organism=organism,
                            description=description,
                            sequence=sequence,
                            header=header,
                        )
                    )
            header = line[1:].strip()
            chunks = []
        else:
            chunks.append(line.strip())
    if header is not None:
        accession, label, organism, description = _parse_fasta_header(header)
        sequence = _normalize_sequence("".join(chunks))
        if sequence:
            records.append(
                SequenceRecord(
                    accession=accession,
                    label=label,
                    organism=organism,
                    description=description,
                    sequence=sequence,
                    header=header,
                )
            )
    return records


def _parse_fasta_header(header: str) -> tuple[str, str, str, str]:
    parts = header.split()
    accession = parts[0] if parts else header[:40]
    label = accession
    organism = ""
    description = header
    if "|OS=" in header:
        description, _, rest = header.partition("OS=")
        organism = rest.split(" OX=")[0].strip()
        description = description.strip()
    return accession, label, organism, description


def _maybe_build_blast_db(fasta_path: Path, collection_name: str, molecule_type: str) -> str | None:
    makeblastdb = shutil.which("makeblastdb")
    if not makeblastdb:
        return None
    dbtype = "prot" if molecule_type.lower().startswith("prot") else "nucl"
    out_prefix = _cache_root() / "blast_db" / collection_name
    cmd = [makeblastdb, "-in", str(fasta_path), "-dbtype", dbtype, "-out", str(out_prefix)]
    try:
        subprocess.run(cmd, check=True, capture_output=True, text=True)
    except Exception:
        return None
    return str(out_prefix)


def _resolve_sequence_collection(collection_name: str) -> dict[str, Any] | None:
    with _connect() as conn:
        if collection_name:
            row = conn.execute(
                "SELECT * FROM sequence_collections WHERE name = ?",
                (collection_name,),
            ).fetchone()
            return dict(row) if row else None
        row = conn.execute(
            "SELECT * FROM sequence_collections ORDER BY created_at DESC LIMIT 1"
        ).fetchone()
        return dict(row) if row else None


def _load_sequences(collection_name: str) -> list[dict[str, Any]]:
    with _connect() as conn:
        rows = conn.execute(
            """
            SELECT accession, label, organism, description, sequence, seq_length, header
            FROM sequences
            WHERE collection_name = ?
            """,
            (collection_name,),
        ).fetchall()
    return [dict(row) for row in rows]


def _run_local_blast(query_sequence: str, collection: dict[str, Any], top_k: int) -> dict[str, Any] | None:
    blast_prefix = collection.get("blast_db_prefix")
    molecule_type = str(collection.get("molecule_type", "protein")).lower()
    if not blast_prefix:
        return None
    blast_binary = shutil.which("blastp" if molecule_type.startswith("prot") else "blastn")
    if not blast_binary:
        return None

    with tempfile.TemporaryDirectory(prefix="hypobioos_local_blast_") as tmp:
        tmp_dir = Path(tmp)
        query_path = tmp_dir / "query.fa"
        out_path = tmp_dir / "hits.tsv"
        query_path.write_text(">query\n" + query_sequence + "\n", encoding="utf-8")
        outfmt = "6 sseqid pident length qstart qend sstart send evalue bitscore"
        cmd = [
            blast_binary,
            "-query",
            str(query_path),
            "-db",
            str(blast_prefix),
            "-outfmt",
            outfmt,
            "-max_target_seqs",
            str(max(top_k, 10)),
            "-out",
            str(out_path),
        ]
        try:
            subprocess.run(cmd, check=True, capture_output=True, text=True)
        except Exception:
            return None

        hits = []
        if out_path.exists():
            for line in out_path.read_text(encoding="utf-8").splitlines():
                fields = line.split("\t")
                if len(fields) != 9:
                    continue
                hits.append(
                    {
                        "accession": fields[0],
                        "identity": round(float(fields[1]) / 100.0, 4),
                        "aligned_length": int(fields[2]),
                        "query_start": int(fields[3]),
                        "query_end": int(fields[4]),
                        "subject_start": int(fields[5]),
                        "subject_end": int(fields[6]),
                        "evalue": fields[7],
                        "bitscore": float(fields[8]),
                    }
                )

    return {
        "method": "local_blast",
        "collection_name": collection["name"],
        "query_length": len(query_sequence),
        "fallback_used": False,
        "hits": hits[:top_k],
    }


def _search_sequences_python(
    query_sequence: str,
    sequences: list[dict[str, Any]],
    *,
    top_k: int,
    min_identity: float,
) -> list[dict[str, Any]]:
    fragment_queries = [query_sequence]
    if len(query_sequence) >= 300:
        fragment_size = max(80, len(query_sequence) // 4)
        fragment_queries = [
            query_sequence[start : start + fragment_size]
            for start in range(0, len(query_sequence) - fragment_size + 1, max(fragment_size // 2, 1))
        ][:4]

    scored = []
    for row in sequences:
        target = row["sequence"]
        best = _best_sequence_match(query_sequence, target)
        fragment_best = best
        if best["identity"] < min_identity and len(fragment_queries) > 1:
            for fragment in fragment_queries:
                candidate = _best_sequence_match(fragment, target)
                if candidate["identity"] > fragment_best["identity"]:
                    fragment_best = candidate | {"fragment_query_length": len(fragment)}
        result = fragment_best
        if result["identity"] < min_identity:
            continue
        scored.append(
            {
                "accession": row.get("accession"),
                "label": row.get("label"),
                "organism": row.get("organism"),
                "description": row.get("description"),
                "identity": round(result["identity"], 4),
                "coverage": round(result["coverage"], 4),
                "subject_start": result["subject_start"],
                "subject_end": result["subject_end"],
                "query_length": result["query_length"],
                "subject_length": len(target),
                "matched_via_fragment": "fragment_query_length" in result,
                "fragment_query_length": result.get("fragment_query_length"),
            }
        )

    scored.sort(key=lambda item: (item["identity"], item["coverage"], -(item["subject_length"])), reverse=True)
    return scored[:top_k]


def _best_sequence_match(query: str, target: str) -> dict[str, Any]:
    if not query or not target:
        return {"identity": 0.0, "coverage": 0.0, "subject_start": 0, "subject_end": 0, "query_length": len(query)}
    if len(query) <= len(target):
        best_identity = 0.0
        best_start = 0
        for start in range(0, len(target) - len(query) + 1):
            window = target[start : start + len(query)]
            matches = sum(1 for a, b in zip(query, window, strict=False) if a == b)
            identity = matches / len(query)
            if identity > best_identity:
                best_identity = identity
                best_start = start
        return {
            "identity": best_identity,
            "coverage": 1.0,
            "subject_start": best_start + 1,
            "subject_end": best_start + len(query),
            "query_length": len(query),
        }

    best = _best_sequence_match(target, query)
    return {
        "identity": best["identity"],
        "coverage": len(target) / len(query),
        "subject_start": 1,
        "subject_end": len(target),
        "query_length": len(query),
    }


def _iter_table_rows(path: Path, *, file_format: str) -> list[dict[str, Any]]:
    suffix = path.suffix.lower()
    resolved_format = file_format
    if resolved_format == "auto":
        if suffix in {".jsonl", ".ndjson"}:
            resolved_format = "jsonl"
        elif suffix == ".json":
            resolved_format = "json"
        elif suffix == ".csv":
            resolved_format = "csv"
        else:
            resolved_format = "tsv"

    if resolved_format == "jsonl":
        return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
    if resolved_format == "json":
        payload = json.loads(path.read_text(encoding="utf-8"))
        if isinstance(payload, list):
            return payload
        raise ValueError("JSON TF binding file must contain a list of row objects.")

    delimiter = "," if resolved_format == "csv" else "\t"
    with path.open("r", encoding="utf-8", newline="") as handle:
        return list(csv.DictReader(handle, delimiter=delimiter))


def _normalize_tf_row(row: dict[str, Any], *, source: str) -> dict[str, Any]:
    lowered = {str(key).lower().strip(): value for key, value in row.items()}

    def pick(*names: str) -> Any:
        for name in names:
            if name in lowered and lowered[name] not in {None, ""}:
                return lowered[name]
        return None

    score = pick("score", "signal", "qvalue", "pvalue", "fold_enrichment")
    try:
        score_value = float(score) if score not in {None, ""} else None
    except Exception:
        score_value = None

    start = pick("start", "chromstart", "peak_start")
    end = pick("end", "chromend", "peak_end")
    try:
        start_value = int(float(start)) if start not in {None, ""} else None
    except Exception:
        start_value = None
    try:
        end_value = int(float(end)) if end not in {None, ""} else None
    except Exception:
        end_value = None

    return {
        "tf_name": _string_or_none(pick("tf_name", "tf", "transcription_factor", "factor", "gene_symbol")),
        "target_gene": _string_or_none(pick("target_gene", "target", "gene", "gene_name", "target_symbol")),
        "cell_type": _string_or_none(pick("cell_type", "cell", "cell line", "cell_line", "tissue")),
        "biosample": _string_or_none(pick("biosample", "sample", "biosample_name")),
        "evidence": _string_or_none(pick("evidence", "assay", "experiment", "peak_id", "dataset")),
        "score": score_value,
        "chrom": _string_or_none(pick("chrom", "chr", "chromosome")),
        "start": start_value,
        "end": end_value,
        "source": _string_or_none(pick("source")) or source,
        "raw": row,
    }


def _needleman_wunsch(
    seq_a: str,
    seq_b: str,
    match_score: int,
    mismatch_score: int,
    gap_score: int,
    *,
    local: bool,
) -> dict[str, Any]:
    rows = len(seq_a) + 1
    cols = len(seq_b) + 1
    score = [[0] * cols for _ in range(rows)]
    trace = [[""] * cols for _ in range(rows)]

    if not local:
        for i in range(1, rows):
            score[i][0] = i * gap_score
            trace[i][0] = "U"
        for j in range(1, cols):
            score[0][j] = j * gap_score
            trace[0][j] = "L"

    best_i = 0
    best_j = 0
    best_score = 0 if local else float("-inf")

    for i in range(1, rows):
        for j in range(1, cols):
            diag = score[i - 1][j - 1] + (match_score if seq_a[i - 1] == seq_b[j - 1] else mismatch_score)
            up = score[i - 1][j] + gap_score
            left = score[i][j - 1] + gap_score
            candidates = [(diag, "D"), (up, "U"), (left, "L")]
            if local:
                candidates.append((0, "0"))
            cell_score, cell_trace = max(candidates, key=lambda item: item[0])
            score[i][j] = cell_score
            trace[i][j] = cell_trace
            if cell_score > best_score:
                best_score = cell_score
                best_i = i
                best_j = j

    if local:
        i, j = best_i, best_j
        final_score = best_score
    else:
        i, j = len(seq_a), len(seq_b)
        final_score = score[i][j]

    aligned_a: list[str] = []
    aligned_b: list[str] = []
    while i > 0 or j > 0:
        direction = trace[i][j]
        if local and (direction == "0" or score[i][j] == 0):
            break
        if direction == "D":
            aligned_a.append(seq_a[i - 1])
            aligned_b.append(seq_b[j - 1])
            i -= 1
            j -= 1
        elif direction == "U":
            aligned_a.append(seq_a[i - 1])
            aligned_b.append("-")
            i -= 1
        elif direction == "L":
            aligned_a.append("-")
            aligned_b.append(seq_b[j - 1])
            j -= 1
        else:
            break

    aligned_a_str = "".join(reversed(aligned_a))
    aligned_b_str = "".join(reversed(aligned_b))
    aligned_pairs = [(a, b) for a, b in zip(aligned_a_str, aligned_b_str, strict=False) if a != "-" and b != "-"]
    matches = sum(1 for a, b in aligned_pairs if a == b)
    identity = matches / len(aligned_pairs) if aligned_pairs else 0.0
    coverage_a = sum(1 for char in aligned_a_str if char != "-") / len(seq_a) if seq_a else 0.0
    coverage_b = sum(1 for char in aligned_b_str if char != "-") / len(seq_b) if seq_b else 0.0

    return {
        "mode": "local" if local else "global",
        "score": final_score,
        "identity": round(identity, 4),
        "coverage_a": round(coverage_a, 4),
        "coverage_b": round(coverage_b, 4),
        "aligned_sequence_a": aligned_a_str,
        "aligned_sequence_b": aligned_b_str,
    }


def _normalize_sequence(sequence: str) -> str:
    return "".join(ch for ch in sequence.upper() if ch.isalpha())


def _string_or_none(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _best_effort_remove_path(path: Path) -> None:
    try:
        if path.exists():
            path.unlink()
    except Exception:
        pass