#!/usr/bin/env python3 """ PPI Skill - Protein-Protein Interaction Analysis 蛋白互作网络分析技能 (基于STRING数据库) Usage: python run.py --gene-ids genes.txt --species ath --prefix Ft.SPL --output ./ppi python run.py --config config.yaml --output ./ppi python run.py --demo --output ./ppi """ import argparse import hashlib import json import logging import math import os import sys import urllib.parse from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any, Callable, Dict, List, Optional try: import requests HAS_REQUESTS = True except ImportError: HAS_REQUESTS = False import pandas as pd import yaml from gfa_core.id_resolution import IdResolutionError, ReferenceAliasIndex from gfa_core.ppi_string_policy import ( STRING_SPECIES_MAP, normalize_gene_ids_for_string, string_taxon_id, unique_preserve_order, ) from gfa_core.string_rate_gate import ( StringRateGate, is_authoritative_string_https_url, string_https_request_identity, ) from gfa_core.string_species_registry import StringSnapshotError, load_string_snapshot logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class StringApiError(RuntimeError): """Typed failure from a STRING transport, HTTP, JSON, or protocol boundary.""" def __init__( self, message: str, *, kind: str, status_code: int | None = None, endpoint: str = "", ) -> None: super().__init__(message) self.kind = kind self.status_code = status_code self.endpoint = endpoint class StringAPI: """STRING API client with stable-version discovery and typed failures.""" BASE_URL = "https://string-db.org/api" CALLER_IDENTITY = "PlantAI" SPECIES_MAP = STRING_SPECIES_MAP def __init__( self, species: str = "ath", *, http_client: Any = None, session: Any = None, stable_api_base: str | None = None, rate_gate: StringRateGate | Any | None = None, rate_gate_path: Path | str | None = None, min_interval: float = 1.0, clock: Callable[[], float] | None = None, sleeper: Callable[[float], None] | None = None, ): self.species = species self.species_id = self.resolve_species_id(species) self._http = http_client or session or (requests if HAS_REQUESTS else None) self.stable_api_base = self._normalize_api_base(stable_api_base or "") gate_options: Dict[str, Any] = { "path": rate_gate_path, "min_interval": min_interval, } if clock is not None: gate_options["clock"] = clock if sleeper is not None: gate_options["sleeper"] = sleeper self.rate_gate = rate_gate or StringRateGate(**gate_options) @staticmethod def resolve_species_id(species: str) -> str: raw = str(species or "").strip() if raw.isdigit(): snapshot_root = Path(__file__).resolve().parents[2] / "data" / "string" / "v12.0" try: snapshot = load_string_snapshot(snapshot_root) except StringSnapshotError as exc: raise ValueError(f"STRING snapshot registry unavailable: {exc}") from exc if raw not in snapshot.registry: raise ValueError(f"STRING TaxID {raw} is not in the fixed STRING snapshot registry") return raw species_id = string_taxon_id(species, "") if not species_id: raise ValueError(f"Unsupported or empty STRING species: {species!r}") return species_id @staticmethod def _normalize_api_base(address: str) -> str: value = str(address or "").strip() if not value: return "" try: parsed = urllib.parse.urlsplit(value) except ValueError as exc: raise StringApiError( "Malformed STRING stable address", kind="protocol", endpoint="version" ) from exc hostname = (parsed.hostname or "").lower() if ( not is_authoritative_string_https_url(value) or parsed.query or parsed.fragment or parsed.path not in {"", "/", "/api", "/api/"} ): raise StringApiError( "Untrusted STRING stable address", kind="protocol", endpoint="version" ) return f"https://{hostname}/api" def _request_json( self, method: str, url: str, *, allow_404: bool = False, **kwargs: Any, ) -> Any: if self._http is None: raise StringApiError( "requests module not available", kind="transport", endpoint=url ) self.rate_gate.wait() try: request_method = getattr(self._http, method.lower()) response = request_method( url, timeout=30, allow_redirects=False, **kwargs ) except Exception as exc: raise StringApiError( f"STRING transport failure: {exc}", kind="transport", endpoint=url ) from exc response_url = getattr(response, "url", None) if string_https_request_identity(response_url) != string_https_request_identity( url ): raise StringApiError( f"STRING response URL does not match request: {response_url!r}", kind="protocol", endpoint=url, ) raw_status = getattr(response, "status_code", None) if isinstance(raw_status, bool) or not isinstance(raw_status, int): raise StringApiError( f"Malformed STRING HTTP status: {raw_status!r}", kind="protocol", endpoint=url, ) status_code = raw_status if allow_404 and status_code == 404: return [] if status_code < 200 or status_code >= 300: raise StringApiError( f"STRING HTTP {status_code}", kind="http", status_code=status_code, endpoint=url, ) try: return response.json() except Exception as exc: raise StringApiError( f"STRING returned invalid JSON: {exc}", kind="json", endpoint=url ) from exc def get_version(self) -> Dict[str, Any]: """Discover and retain the stable API address for this STRING version.""" payload = self._request_json( "GET", f"{self.BASE_URL}/json/version", params={"caller_identity": self.CALLER_IDENTITY}, ) if isinstance(payload, list) and len(payload) == 1: version = payload[0] elif isinstance(payload, dict): version = payload else: raise StringApiError( "STRING version response must be one object", kind="protocol", endpoint="version", ) if not isinstance(version, dict): raise StringApiError( "STRING version response contains a non-object", kind="protocol", endpoint="version", ) stable_address = version.get("string_stable_address") or version.get( "stable_address" ) if not isinstance(stable_address, str) or not stable_address.strip(): raise StringApiError( "STRING version response has no stable address", kind="protocol", endpoint="version", ) self.stable_api_base = self._normalize_api_base(stable_address) return dict(version) def _stable_base(self) -> str: if not self.stable_api_base: self.get_version() return self.stable_api_base @classmethod def validate_mapping_rows( cls, payload: Any, identifiers: List[str], species_id: str, ) -> List[Dict[str, Any]]: if not isinstance(payload, list): raise StringApiError( "STRING mapping response must be a list", kind="protocol" ) seen_indices = set() validated = [] expected_taxon = int(species_id) for row in payload: if not isinstance(row, dict): raise StringApiError( "STRING mapping row must be an object", kind="protocol" ) query_index = row.get("queryIndex") if ( isinstance(query_index, bool) or not isinstance(query_index, int) or query_index < 0 or query_index >= len(identifiers) or query_index in seen_indices ): raise StringApiError( f"Invalid STRING queryIndex: {query_index!r}", kind="protocol" ) seen_indices.add(query_index) taxon = row.get("ncbiTaxonId") parsed_taxon = cls._validated_taxon(taxon) if parsed_taxon != expected_taxon: raise StringApiError( f"STRING mapping taxon {parsed_taxon} != {expected_taxon}", kind="protocol", ) cls._validated_string_id(row.get("stringId"), parsed_taxon) if ( not isinstance(row.get("queryItem"), str) or row["queryItem"] != identifiers[query_index] or not isinstance(row.get("stringId"), str) or not row["stringId"].strip() or not isinstance(row.get("preferredName"), str) or not row["preferredName"].strip() ): raise StringApiError("Malformed STRING mapping row", kind="protocol") validated.append(dict(row)) return sorted(validated, key=lambda row: row["queryIndex"]) @staticmethod def _validated_taxon(value: Any) -> int: if isinstance(value, bool): raise StringApiError(f"Invalid STRING taxon: {value!r}", kind="protocol") if isinstance(value, int): return value if isinstance(value, str) and value.isdigit(): return int(value) raise StringApiError(f"Invalid STRING taxon: {value!r}", kind="protocol") @staticmethod def _validated_string_id(value: Any, taxon_id: int) -> str: if not isinstance(value, str) or value != value.strip(): raise StringApiError( f"Invalid STRING identifier: {value!r}", kind="protocol" ) prefix, separator, entity = value.partition(".") if ( separator != "." or prefix != str(taxon_id) or not entity or any(character.isspace() for character in entity) ): raise StringApiError( f"Invalid STRING identifier for taxon {taxon_id}: {value!r}", kind="protocol", ) return value def get_string_ids(self, identifiers: List[str]) -> List[Dict[str, Any]]: """Resolve candidate identifiers, attributing rows only by queryIndex.""" self._validate_identifiers(identifiers) url = f"{self._stable_base()}/json/get_string_ids" payload = self._request_json( "POST", url, allow_404=True, data={ "identifiers": "\r\n".join(identifiers), "species": self.species_id, "echo_query": 1, "caller_identity": self.CALLER_IDENTITY, }, ) return self.validate_mapping_rows(payload, identifiers, self.species_id) @staticmethod def _validate_identifiers(identifiers: List[str]) -> None: if not isinstance(identifiers, list) or not identifiers: raise ValueError("identifiers must be a non-empty list") if any(not isinstance(item, str) or not item.strip() for item in identifiers): raise ValueError("identifiers must contain non-empty strings") def get_network( self, proteins: List[str], required_score: int = 200, network_type: str = "functional", ) -> List[Dict[str, Any]]: """Return a validated STRING network; an empty list is scientifically valid.""" self._validate_identifiers(proteins) if ( isinstance(required_score, bool) or not isinstance(required_score, int) or not 0 <= required_score <= 1000 ): raise ValueError("required_score must be an integer from 0 to 1000") if network_type != "functional": raise ValueError("network_type is fixed to 'functional'") selected_taxon = int(self.species_id) for protein in proteins: self._validated_string_id(protein, selected_taxon) submitted_ids = set(proteins) url = f"{self._stable_base()}/json/network" payload = self._request_json( "POST", url, data={ "identifiers": "\r\n".join(proteins), "species": self.species_id, "required_score": required_score, "network_type": "functional", "add_nodes": 0, "caller_identity": self.CALLER_IDENTITY, }, ) if not isinstance(payload, list): raise StringApiError( "STRING network response must be a list", kind="protocol", endpoint=url, ) for row in payload: if not isinstance(row, dict): raise StringApiError( "Malformed STRING network row", kind="protocol", endpoint=url ) parsed_taxon = self._validated_taxon(row.get("ncbiTaxonId")) preferred_a = row.get("preferredName_A") preferred_b = row.get("preferredName_B") score = row.get("score") if ( parsed_taxon != int(self.species_id) or not isinstance(preferred_a, str) or not preferred_a.strip() or not isinstance(preferred_b, str) or not preferred_b.strip() or isinstance(score, bool) or not isinstance(score, (int, float)) or not math.isfinite(score) or not 0 <= score <= 1 ): raise StringApiError( "Malformed STRING network row", kind="protocol", endpoint=url ) string_id_a = self._validated_string_id( row.get("stringId_A"), parsed_taxon ) string_id_b = self._validated_string_id( row.get("stringId_B"), parsed_taxon ) if ( string_id_a not in submitted_ids or string_id_b not in submitted_ids or score < required_score / 1000 ): raise StringApiError( "STRING network row contradicts submitted identifiers or score threshold", kind="protocol", endpoint=url, ) return payload def generate_string_url(proteins: List[str], species: str = "3702", required_score: int = 200) -> str: """生成STRING网页链接 Args: proteins: 蛋白ID列表 species: STRING物种ID (默认3702=拟南芥) Returns: STRING网络查看URL """ base_url = "https://string-db.org/cgi/network" # 使用实际换行符，urllib.parse.urlencode会自动编码 identifiers = "\r\n".join(proteins) params = { "identifiers": identifiers, "species": species, "required_score": str(required_score), "network_type": "functional" } return f"{base_url}?{urllib.parse.urlencode(params)}" @dataclass(frozen=True, slots=True) class PpiQueryOrigin: canonical_target_id: str ortholog_reference_id: str raw_query_id: str relationship_source: str @dataclass(frozen=True, slots=True) class PpiResolutionRow: canonical_target_id: str ortholog_reference_id: str raw_query_id: str candidate_gene_id: str selected_query_id: str string_id: str string_preferred_name: str mapping_status: str method: str evidence_source: str relationship_source: str species: str taxon_id: str reason: str @dataclass(slots=True) class StringResolutionResult: selected_string_ids: List[str] rows: List[PpiResolutionRow] metrics: Dict[str, Any] status: str = "success" issue_code: str = "" @dataclass(frozen=True, slots=True) class _ResolutionCandidate: value: str namespace: str method: str evidence_source: str def parse_ortholog_mapping(mapping_file: Path) -> List[PpiQueryOrigin]: """Parse every canonical-to-reference relationship without rewriting IDs.""" mapping: List[PpiQueryOrigin] = [] if not mapping_file.exists(): return mapping with open(mapping_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue parts = line.split("\t") if len(parts) >= 2: source_id = parts[0].strip() target_id = parts[1].strip() if source_id and target_id: mapping.append( PpiQueryOrigin( canonical_target_id=source_id, ortholog_reference_id=target_id, raw_query_id=target_id, relationship_source=str(mapping_file), ) ) return mapping class PpiSkill: """PPI分析技能主类""" def __init__( self, config_path: Optional[str] = None, *, string_api_factory: Callable[[str], Any] | None = None, ): self.config = self._load_config(config_path) self.checksums = {} self.start_time = None self.end_time = None self.ortholog_mapping: List[PpiQueryOrigin] | None = None self.string_api_factory = string_api_factory or StringAPI def _load_config(self, config_path: Optional[str]) -> Dict: """加载配置文件""" if config_path and os.path.exists(config_path): with open(config_path) as f: return yaml.safe_load(f) default_config = Path(__file__).parent / "config.yaml" with open(default_config) as f: return yaml.safe_load(f) def _compute_checksum(self, filepath: Path) -> str: """计算文件SHA256校验""" sha256 = hashlib.sha256() with open(filepath, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): sha256.update(chunk) return sha256.hexdigest() def _record_file(self, filepath: Path, file_type: str = "output") -> Dict: """记录文件信息""" if filepath.exists(): return { "path": str(filepath), "sha256": self._compute_checksum(filepath), "size": filepath.stat().st_size, "type": file_type } return {} def load_gene_ids(self, gene_file: Path) -> List[str]: """加载基因ID列表""" gene_ids = [] with open(gene_file, 'r') as f: for line in f: gene_id = line.strip() if gene_id: gene_ids.append(gene_id) return gene_ids def load_ortholog_mapping(self, mapping_file: Path) -> List[PpiQueryOrigin]: """加载同源基因映射""" self.ortholog_mapping = parse_ortholog_mapping(mapping_file) logger.info(f"Loaded {len(self.ortholog_mapping)} ortholog mappings from {mapping_file}") return self.ortholog_mapping @staticmethod def _query_candidates( raw_query_id: str, normalization_species_label: str, alias_index: ReferenceAliasIndex | None, ) -> List[_ResolutionCandidate]: raw = str(raw_query_id or "").strip() candidates: List[_ResolutionCandidate] = [] def add(value: str, namespace: str, method: str, evidence_source: str) -> None: token = str(value or "").strip() if token and all(candidate.value != token for candidate in candidates): candidates.append( _ResolutionCandidate(token, namespace, method, evidence_source) ) add(raw, "raw", "raw_query", "raw_query") normalized = normalize_gene_ids_for_string([raw], normalization_species_label) if normalized: add( normalized[0], "normalized", f"species_rule:{normalization_species_label}", "ppi_string_policy", ) if "." in raw and raw.rsplit(".", 1)[1].isdigit(): add( raw.rsplit(".", 1)[0], "versionless", "versionless_lookup_candidate", "raw_query_id", ) if alias_index is not None: for candidate in alias_index.candidates(raw): add( candidate.value, candidate.namespace, candidate.method, candidate.evidence_source, ) return candidates def resolve_query_ids( self, origins: List[PpiQueryOrigin], species: str, id_map_file: Path | None, *, api: Any = None, canonical_target_ids: List[str] | None = None, normalization_species_label: str = "", ) -> StringResolutionResult: """Resolve relationship queries without collapsing their provenance rows.""" species_id = StringAPI.resolve_species_id(species) if not isinstance(origins, list) or any( not isinstance(item, PpiQueryOrigin) for item in origins ): raise ValueError("origins must be a list of PpiQueryOrigin") alias_index = ( ReferenceAliasIndex.from_tsv( Path(id_map_file), str(normalization_species_label or species).strip(), ) if id_map_file is not None else None ) unique_queries = unique_preserve_order( [origin.raw_query_id for origin in origins if origin.raw_query_id] ) candidates_by_query: Dict[str, List[_ResolutionCandidate]] = {} query_errors: Dict[str, IdResolutionError] = {} normalization_label = str(normalization_species_label or species).strip() for raw_query in unique_queries: try: candidates_by_query[raw_query] = self._query_candidates( raw_query, normalization_label, alias_index ) except IdResolutionError as exc: candidates_by_query[raw_query] = [] query_errors[raw_query] = exc probes = unique_preserve_order( [ candidate.value for raw_query in unique_queries for candidate in candidates_by_query[raw_query] ] ) string_api = api or self.string_api_factory(species) mapping_rows = string_api.get_string_ids(probes) if probes else [] validated_rows = StringAPI.validate_mapping_rows( mapping_rows, probes, species_id ) mapped_by_probe = { probes[row["queryIndex"]]: row for row in validated_rows } query_results: Dict[str, Dict[str, Any]] = {} for raw_query in unique_queries: candidates = candidates_by_query[raw_query] successful = [ (candidate, mapped_by_probe[candidate.value]) for candidate in candidates if candidate.value in mapped_by_probe ] string_ids = unique_preserve_order( [row["stringId"] for _candidate, row in successful] ) if raw_query in query_errors: query_results[raw_query] = { "status": "ambiguous", "reason": query_errors[raw_query].code, "string_id": "", "preferred_name": "", "selected_candidate": None, "raw_resolved": False, "geneid_fallback": False, } elif not successful: query_results[raw_query] = { "status": "unresolved", "reason": "no_string_mapping", "string_id": "", "preferred_name": "", "selected_candidate": None, "raw_resolved": False, "geneid_fallback": False, } elif len(string_ids) > 1: query_results[raw_query] = { "status": "ambiguous", "reason": "candidate_string_ids_conflict", "string_id": "", "preferred_name": "", "selected_candidate": None, "raw_resolved": False, "geneid_fallback": False, } else: selected_candidate, selected_row = successful[0] raw_resolved = any( candidate.namespace in {"raw", "normalized", "versionless"} for candidate, _row in successful ) geneid_resolved = any( candidate.namespace == "ncbi_gene_id" for candidate, _row in successful ) query_results[raw_query] = { "status": "resolved", "reason": "candidates_converged", "string_id": string_ids[0], "preferred_name": str(selected_row.get("preferredName") or ""), "selected_candidate": selected_candidate, "raw_resolved": raw_resolved, "geneid_fallback": geneid_resolved and not raw_resolved, } rows = [ self._resolution_row(origin, query_results[origin.raw_query_id], species) for origin in origins ] rows = self._exclude_conflicting_orthologs(rows) selected_string_ids = unique_preserve_order( [row.string_id for row in rows if row.mapping_status == "resolved"] ) metrics = self._resolution_metrics( origins, rows, unique_queries, query_results, selected_string_ids, canonical_target_ids=canonical_target_ids, ) if selected_string_ids: return StringResolutionResult(selected_string_ids, rows, metrics) return StringResolutionResult( selected_string_ids, rows, metrics, status="id_resolution_failure", issue_code="ppi_string_id_resolution_failed", ) @staticmethod def _resolution_row( origin: PpiQueryOrigin, query_result: Dict[str, Any], species: str, ) -> PpiResolutionRow: selected = query_result.get("selected_candidate") return PpiResolutionRow( canonical_target_id=origin.canonical_target_id, ortholog_reference_id=origin.ortholog_reference_id, raw_query_id=origin.raw_query_id, candidate_gene_id=( selected.value if selected is not None and selected.namespace in {"ncbi_gene_id", "gene_symbol"} else "" ), selected_query_id=selected.value if selected is not None else "", string_id=query_result["string_id"], string_preferred_name=query_result["preferred_name"], mapping_status=query_result["status"], method=selected.method if selected is not None else "", evidence_source=( ( "ortholog_relationship" if origin.ortholog_reference_id else "canonical_member_input" ) if selected is not None and selected.namespace == "raw" else selected.evidence_source if selected is not None else origin.relationship_source ), relationship_source=origin.relationship_source, species=species, taxon_id=StringAPI.resolve_species_id(species), reason=query_result["reason"], ) @staticmethod def _exclude_conflicting_orthologs( rows: List[PpiResolutionRow], ) -> List[PpiResolutionRow]: rows_by_canonical: Dict[str, List[PpiResolutionRow]] = {} for row in rows: rows_by_canonical.setdefault(row.canonical_target_id, []).append(row) conflicting = set() for canonical, canonical_rows in rows_by_canonical.items(): references = {row.raw_query_id for row in canonical_rows} if len(references) < 2: continue string_ids = { row.string_id for row in canonical_rows if row.mapping_status == "resolved" and row.string_id } if ( len(string_ids) != 1 or any(row.mapping_status != "resolved" for row in canonical_rows) ): conflicting.add(canonical) return [ PpiResolutionRow( **{ **{ field: getattr(row, field) for field in row.__dataclass_fields__ }, "candidate_gene_id": "", "selected_query_id": "", "string_id": "", "string_preferred_name": "", "mapping_status": "ambiguous_ortholog", "method": "", "reason": "ortholog_references_do_not_converge", } ) if row.canonical_target_id in conflicting else row for row in rows ] @staticmethod def _resolution_metrics( origins: List[PpiQueryOrigin], rows: List[PpiResolutionRow], unique_queries: List[str], query_results: Dict[str, Dict[str, Any]], selected_string_ids: List[str], *, canonical_target_ids: List[str] | None = None, ) -> Dict[str, Any]: input_members = unique_preserve_order( list(canonical_target_ids or []) + [origin.canonical_target_id for origin in origins] ) resolved_queries = sum( result["status"] == "resolved" for result in query_results.values() ) states_by_member: Dict[str, set[str]] = {} for row in rows: states_by_member.setdefault(row.canonical_target_id, set()).add( row.mapping_status ) resolved_members = sum(states == {"resolved"} for states in states_by_member.values()) ambiguous_members = sum( bool(states.intersection({"ambiguous", "ambiguous_ortholog"})) for states in states_by_member.values() ) unresolved_members = len(input_members) - resolved_members - ambiguous_members input_count = len(input_members) query_count = len(unique_queries) probe_status = ( "complete" if query_count and resolved_queries == query_count else "partial" if resolved_queries else "failed" ) return { "input_member_count": input_count, "relationship_count": len(origins), "unique_reference_query_count": query_count, "raw_resolved_query_count": sum( bool(result["raw_resolved"]) for result in query_results.values() ), "geneid_fallback_resolved_query_count": sum( bool(result["geneid_fallback"]) for result in query_results.values() ), "resolved_reference_query_count": resolved_queries, "selected_string_query_count": len(selected_string_ids), "resolved_member_count": resolved_members, "unresolved_member_count": unresolved_members, "ambiguous_member_count": ambiguous_members, "member_mapping_rate": round(resolved_members / input_count, 6) if input_count else 0, "query_mapping_rate": round(resolved_queries / query_count, 6) if query_count else 0, "probe_status": probe_status, } def write_string_resolution_report( self, resolution: StringResolutionResult, output_dir: Path, prefix: str, ) -> tuple[Path, Dict[str, Any]]: """Serialize one deterministic summary row per input relationship.""" columns = [ "canonical_target_id", "ortholog_reference_id", "raw_query_id", "candidate_gene_id", "selected_query_id", "string_id", "string_preferred_name", "mapping_status", "method", "evidence_source", "relationship_source", "species", "taxon_id", "reason", ] output_file = output_dir / f"{prefix}_id_resolution.tsv" with output_file.open("w", encoding="utf-8") as handle: handle.write("\t".join(columns) + "\n") for row in resolution.rows: values = [ str(getattr(row, column)).replace("\t", " ").replace("\n", " ") for column in columns ] handle.write("\t".join(values) + "\n") return output_file, dict(resolution.metrics) def analyze_network( self, species: str, required_score: int = 200, *, origins: List[PpiQueryOrigin], canonical_target_ids: List[str], id_map_file: Path | None = None, normalization_species_label: str = "", ) -> Dict: """分析互作网络 Returns: Dict with keys: interactions, stats, string_url """ api = self.string_api_factory(species) resolution = self.resolve_query_ids( origins, species, id_map_file, api=api, canonical_target_ids=canonical_target_ids, normalization_species_label=normalization_species_label, ) network_query_ids = resolution.selected_string_ids if resolution.status == "id_resolution_failure": return { "status": "id_resolution_failure", "issue_code": resolution.issue_code, "interactions": [], "stats": {}, "string_url": "", "id_resolution": resolution, } logger.info( f"Querying STRING for {len(network_query_ids)} genes (species: {species})" ) network_data = api.get_network(network_query_ids, required_score) # 生成STRING网页链接 (无论是否有网络数据都生成) string_url = generate_string_url( network_query_ids, api.species_id, required_score ) logger.info(f"STRING URL: {string_url}") if not network_data: logger.info("STRING returned a valid empty network") return { "interactions": [], "stats": { "input_genes": len(network_query_ids), "genes_in_network": 0, "total_interactions": 0, "avg_score": 0, "high_confidence": 0, "medium_confidence": 0, "low_confidence": 0, "network_status": "empty_network", }, "string_url": string_url, "id_resolution": resolution, } # 解析互作数据 interactions = [] proteins_in_network = set() for interaction in network_data: if isinstance(interaction, dict): p1 = interaction.get("preferredName_A", "") p2 = interaction.get("preferredName_B", "") score = interaction.get("score", 0) interactions.append({ "protein1": p1, "protein2": p2, "score": score }) proteins_in_network.add(p1) proteins_in_network.add(p2) # 统计 stats = { "input_genes": len(network_query_ids), "genes_in_network": len(proteins_in_network), "total_interactions": len(interactions), "avg_score": sum(i["score"] for i in interactions) / len(interactions) if interactions else 0, "high_confidence": sum(1 for i in interactions if i["score"] >= 0.7), "medium_confidence": sum(1 for i in interactions if 0.4 <= i["score"] < 0.7), "low_confidence": sum(1 for i in interactions if i["score"] < 0.4), "network_status": "network_detected", } logger.info(f"Network: {stats['genes_in_network']}/{stats['input_genes']} genes, " f"{stats['total_interactions']} interactions") return { "interactions": interactions, "stats": stats, "string_url": string_url, "id_resolution": resolution, } def save_network(self, interactions: List[Dict], output_file: Path) -> Path: """保存网络数据""" df = pd.DataFrame(interactions, columns=["protein1", "protein2", "score"]) df.to_csv(output_file, sep='\t', index=False) logger.info(f"Network saved: {output_file}") return output_file def run( self, gene_ids_file: Path, species: str, prefix: str, output_dir: Path, required_score: int = 200, mapping_file: Optional[Path] = None, id_map_file: Optional[Path] = None, normalization_species_label: str = "", ppi_provenance: Optional[Dict[str, Any]] = None, ) -> Dict: """运行PPI分析 Args: gene_ids_file: 基因ID列表文件 species: 目标物种代码 (用于STRING查询) prefix: 输出前缀 output_dir: 输出目录 required_score: 置信度阈值 mapping_file: 同源映射文件 (可选，用于非模式植物) id_map_file: 当前任务/物种的本地ID证据文件 (可选) normalization_species_label: 仅用于本地ID规范化规则，不作为STRING查询身份 """ self.start_time = datetime.now() logger.info(f"Starting PPI analysis for {prefix}") logger.info(f"Input: {gene_ids_file}") logger.info(f"Species: {species}") logger.info(f"Output: {output_dir}") # 创建输出目录 output_dir.mkdir(parents=True, exist_ok=True) # 加载基因ID gene_ids = self.load_gene_ids(gene_ids_file) logger.info(f"Loaded {len(gene_ids)} gene IDs") if len(gene_ids) == 0: logger.error("No gene IDs loaded") return self._generate_failed_manifest( prefix, output_dir, "No gene IDs", ppi_provenance=ppi_provenance ) # 加载同源映射 (如果有) ortholog_mode = mapping_file is not None self.ortholog_mapping = None if ortholog_mode: if not mapping_file.exists(): return self._generate_failed_manifest( prefix, output_dir, f"Ortholog mapping file does not exist: {mapping_file}", ppi_provenance=ppi_provenance, ) self.load_ortholog_mapping(mapping_file) if ortholog_mode: requested_members = set(gene_ids) origins = [ origin for origin in (self.ortholog_mapping or []) if origin.canonical_target_id in requested_members ] mapped_members = {origin.canonical_target_id for origin in origins} if len(mapped_members) < len(requested_members): logger.warning( "%s/%s canonical members have no Ortholog relationship", len(requested_members) - len(mapped_members), len(requested_members), ) else: origins = [ PpiQueryOrigin( canonical_target_id=gene_id, ortholog_reference_id="", raw_query_id=gene_id, relationship_source=str(gene_ids_file), ) for gene_id in gene_ids ] result = self.analyze_network( species, required_score, origins=origins, id_map_file=id_map_file, canonical_target_ids=gene_ids, normalization_species_label=normalization_species_label, ) resolution = result["id_resolution"] id_resolution_file, id_resolution_metrics = ( self.write_string_resolution_report(resolution, output_dir, prefix) ) if result.get("status") == "id_resolution_failure": return self._generate_failed_manifest( prefix, output_dir, "No identifiers resolved unambiguously in STRING", issue_code=str(result.get("issue_code") or ""), metrics={"id_resolution": id_resolution_metrics}, artifacts={"id_resolution": self._record_file(id_resolution_file, "tsv")}, ppi_provenance=ppi_provenance, ) query_ids = resolution.selected_string_ids interactions = result["interactions"] stats = result["stats"] string_url = result.get("string_url", "") # 保存网络数据 network_file = output_dir / f"{prefix}_network.tsv" self.save_network(interactions, network_file) # 保存STRING URL到文件 if string_url: url_file = output_dir / f"{prefix}_url.txt" with open(url_file, 'w') as f: f.write("# STRING Protein Interaction Network\n\n") f.write(f"**Query**: {prefix}\n") f.write(f"**Species**: {species}\n") f.write(f"**Genes**: {len(query_ids)}\n\n") f.write(f"View online: {string_url}\n") logger.info(f"STRING URL saved: {url_file}") # 生成报告 report_path = self._generate_report(stats, prefix, output_dir, species, string_url) # 生成manifest manifest = self._generate_manifest( gene_ids_file, species, prefix, output_dir, stats, network_file, report_path, string_url, id_resolution_file=id_resolution_file, id_resolution_metrics=id_resolution_metrics, normalization_species_label=normalization_species_label, ppi_provenance=ppi_provenance, ) self.end_time = datetime.now() elapsed = (self.end_time - self.start_time).total_seconds() logger.info(f"PPI analysis completed in {elapsed:.1f}s") return manifest def _generate_report(self, stats: Dict, prefix: str, output_dir: Path, species: str, string_url: str = "") -> Path: """生成Markdown报告""" report_path = output_dir / f"{prefix}_report.md" with open(report_path, 'w') as f: f.write(f"# PPI Network Analysis Report - {prefix}\n\n") f.write(f"**Generated**: {datetime.now().isoformat()}\n") f.write(f"**Species**: {species}\n") f.write(f"**Data Source**: STRING Database\n\n") f.write("## Network Statistics\n\n") f.write(f"- **Input Genes**: {stats.get('input_genes', 0)}\n") f.write(f"- **Genes in Network**: {stats.get('genes_in_network', 0)}\n") f.write(f"- **Total Interactions**: {stats.get('total_interactions', 0)}\n") f.write(f"- **Average Score**: {stats.get('avg_score', 0):.3f}\n") f.write(f"- **High Confidence (≥0.7)**: {stats.get('high_confidence', 0)}\n") f.write(f"- **Medium Confidence (0.4-0.7)**: {stats.get('medium_confidence', 0)}\n") f.write(f"- **Low Confidence (<0.4)**: {stats.get('low_confidence', 0)}\n\n") if string_url: f.write("## STRING Network\n\n") f.write(f"[View Interactive Network]({string_url})\n\n") f.write("## Output Files\n\n") f.write(f"- `{prefix}_network.tsv` - Interaction network data\n") f.write(f"- `{prefix}_report.md` - This report\n") return report_path def _generate_manifest(self, input_file: Path, species: str, prefix: str, output_dir: Path, stats: Dict, network_file: Path, report: Path, string_url: str = "", id_resolution_file: Optional[Path] = None, id_resolution_metrics: Optional[Dict] = None, normalization_species_label: str = "", ppi_provenance: Optional[Dict[str, Any]] = None) -> Dict: """生成manifest""" self.end_time = datetime.now() elapsed = (self.end_time - self.start_time).total_seconds() artifacts = {} if network_file and network_file.exists(): artifacts["network"] = self._record_file(network_file, "tsv") if report and report.exists(): artifacts["report"] = self._record_file(report, "markdown") if id_resolution_file and id_resolution_file.exists(): artifacts["id_resolution"] = self._record_file(id_resolution_file, "tsv") # 添加URL文件到artifacts url_file = output_dir / f"{prefix}_url.txt" if url_file.exists(): artifacts["string_url"] = self._record_file(url_file, "txt") manifest = { "manifest_version": "1.0", "stage": "ppi", "status": "success", "prefix": prefix, "timestamp": self.end_time.isoformat(), "elapsed_sec": round(elapsed, 2), "input": {"gene_ids": str(input_file), "species": species}, "parameters": { "string_taxon_id": StringAPI.resolve_species_id(species), "normalization_species_label": str(normalization_species_label or species).strip(), }, "metrics": { **stats, "id_resolution": id_resolution_metrics or {}, }, "artifacts": artifacts, "string_url": string_url, "ppi_provenance": dict(ppi_provenance or {}), } manifest_path = output_dir / f"{prefix}.manifest.json" with open(manifest_path, 'w') as f: json.dump(manifest, f, indent=2) manifest["manifest_path"] = str(manifest_path) return manifest def _generate_failed_manifest( self, prefix: str, output_dir: Path, message: str, *, issue_code: str = "ppi_execution_failed", metrics: Optional[Dict[str, Any]] = None, artifacts: Optional[Dict[str, Any]] = None, ppi_provenance: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: output_dir.mkdir(parents=True, exist_ok=True) manifest = { "manifest_version": "1.0", "stage": "ppi", "status": "failed", "prefix": prefix, "timestamp": datetime.now().isoformat(), "message": message, "issue_code": issue_code, "metrics": dict(metrics or {}), "artifacts": dict(artifacts or {}), "ppi_provenance": dict(ppi_provenance or {}), } manifest_path = output_dir / f"{prefix}.manifest.json" manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8") return {**manifest, "manifest_path": str(manifest_path)} def main(): parser = argparse.ArgumentParser(description="PPI Skill - Protein-Protein Interaction Analysis") parser.add_argument("-i", "--gene-ids", help="Gene ID list file") parser.add_argument("-s", "--species", default="ath", help="Species code (ath/osa/zma)") parser.add_argument( "--normalization-species-label", default="", help="Local species label used only for gene-ID normalization rules", ) parser.add_argument("-p", "--prefix", required=True, help="Output prefix") parser.add_argument("-o", "--output", required=True, help="Output directory") parser.add_argument("--score", type=int, default=200, help="Confidence threshold (0-1000)") parser.add_argument("--mapping", help="Ortholog mapping file (for non-model species)") parser.add_argument("--id-map", help="Species-matched local PEP/CDS ID evidence TSV") parser.add_argument("--provenance-json", help="Authoritative PPI provenance JSON from preflight/backend") parser.add_argument("--config", help="Config file path") parser.add_argument("--demo", action="store_true", help="Demo mode") args = parser.parse_args() # Demo模式 if args.demo: output_path = Path(args.output) output_path.mkdir(parents=True, exist_ok=True) demo_genes = output_path / "demo_genes.txt" with open(demo_genes, 'w') as f: f.write("AT1G01010\nAT1G01020\nAT1G01030\n") args.gene_ids = str(demo_genes) if not args.gene_ids: parser.error("--gene-ids is required (or use --demo)") # 运行技能 skill = PpiSkill(config_path=args.config) provenance = {} if args.provenance_json: provenance = json.loads(Path(args.provenance_json).read_text(encoding="utf-8")) try: result = skill.run( gene_ids_file=Path(args.gene_ids), species=args.species, prefix=args.prefix, output_dir=Path(args.output), required_score=args.score, mapping_file=Path(args.mapping) if args.mapping else None, id_map_file=Path(args.id_map) if args.id_map else None, normalization_species_label=args.normalization_species_label, ppi_provenance=provenance, ) except Exception as exc: result = skill._generate_failed_manifest( args.prefix, Path(args.output), str(exc), ppi_provenance=provenance ) if result.get("status") == "failed": print(f"\n❌ PPI analysis failed: {result.get('message')}") sys.exit(1) print(f"\n✅ PPI analysis completed!") print(f" Manifest: {result.get('manifest_path', 'N/A')}") print(f" Genes in network: {result['metrics'].get('genes_in_network', 0)}/{result['metrics'].get('input_genes', 0)}") print(f" Interactions: {result['metrics'].get('total_interactions', 0)}") sys.exit(0) if __name__ == "__main__": main()