Create biosecurity.py

genesis/biosecurity.py

@@ -1,105 +1,70 @@
 # genesis/biosecurity.py
 import os
 import requests
+from typing import Dict, Any
 from datetime import datetime
-from typing import Dict, Any, List
+
+from .providers import pubmed_fallback_search, run_deepseek_summary
 
 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
-BIOPORTAL_API_KEY = os.getenv("BIOPORTAL_API_KEY")
-UMLS_API_KEY = os.getenv("UMLS_API_KEY")
 NCBI_API_KEY = os.getenv("NCBI_API_KEY")
 NCBI_EMAIL = os.getenv("NCBI_EMAIL")
 
-def search_pubmed_recent(query: str, max_results: int = 5) -> List[Dict[str, str]]:
-    """Fetch recent PubMed papers for biosecurity context."""
-    try:
-        url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi"
-        params = {
-            "db": "pubmed",
-            "term": query,
-            "retmax": max_results,
-            "sort": "date",
-            "retmode": "json",
-            "api_key": NCBI_API_KEY,
-            "email": NCBI_EMAIL
-        }
-        r = requests.get(url, params=params, timeout=15)
-        r.raise_for_status()
-        ids = r.json().get("esearchresult", {}).get("idlist", [])
-        
-        papers = []
-        if ids:
-            fetch_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi"
-            fetch_params = {
-                "db": "pubmed",
-                "id": ",".join(ids),
-                "retmode": "json",
-                "api_key": NCBI_API_KEY,
-                "email": NCBI_EMAIL
-            }
-            fr = requests.get(fetch_url, params=fetch_params, timeout=15)
-            fr.raise_for_status()
-            summaries = fr.json().get("result", {})
-            for pmid in ids:
-                if pmid in summaries:
-                    papers.append({
-                        "title": summaries[pmid].get("title", ""),
-                        "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/"
-                    })
-        return papers
-    except Exception as e:
-        print(f"[PubMed Error] {e}")
-        return []
+RISK_KEYWORDS = {
+    "low": ["harmless", "biosafety level 1", "safe", "non-pathogenic"],
+    "medium": ["biosafety level 2", "BSL-2", "infectious", "containment"],
+    "high": ["BSL-3", "BSL-4", "pandemic potential", "gain-of-function", "biosecurity concern"]
+}
+
+def score_biosecurity_risk(text: str) -> int:
+    """Score risk based on keywords in AI report."""
+    text_lower = text.lower()
+    score = 0
+    for word in RISK_KEYWORDS["low"]:
+        if word in text_lower:
+            score += 10
+    for word in RISK_KEYWORDS["medium"]:
+        if word in text_lower:
+            score += 25
+    for word in RISK_KEYWORDS["high"]:
+        if word in text_lower:
+            score += 50
+    return min(score, 100)  # cap at 100
 
-def ai_biosecurity_assessment(entity: str) -> Dict[str, Any]:
-    """Run AI-powered biosecurity risk assessment."""
-    import openai
-    openai.api_key = OPENAI_API_KEY
+def run_biosecurity_scan(entity: str) -> Dict[str, Any]:
+    """
+    Run AI-powered biosecurity risk scan for a given biological entity.
+    Includes AI assessment + PubMed literature.
+    """
     try:
+        # AI assessment
         prompt = f"""
-        You are a synthetic biology biosecurity officer.
-        Assess the biosecurity risk of the following entity: {entity}.
-        
-        Consider:
-        - Is it a known dangerous pathogen, toxin, or dual-use technology?
-        - Potential misuse (bioterrorism, lab escape)
-        - Regulatory oversight and biosafety levels
-        - Recent trends in research or weaponization
-        - Ethical concerns
-        
-        Return:
-        - Risk Score (0-100)
-        - Category (Low, Medium, High)
-        - Reasons
-        - Recommended Actions
+        You are a synthetic biology & biosecurity expert.
+        Assess the potential biosecurity risks of the following entity: {entity}.
+        Classify its biosafety level, potential misuse, regulatory concerns, and safe handling guidelines.
+        Respond with detailed analysis.
         """
-        response = openai.ChatCompletion.create(
-            model="gpt-4o-mini",
-            messages=[{"role": "user", "content": prompt}],
-            temperature=0.2
-        )
-        return {"ai_report": response.choices[0].message["content"]}
-    except Exception as e:
-        print(f"[OpenAI Error] {e}")
-        return {"ai_report": "AI risk analysis unavailable."}
-
-def run_biosecurity_scan(entity: str) -> Dict[str, Any]:
-    """Main function to scan biosecurity risks."""
-    # Step 1: AI assessment
-    ai_results = ai_biosecurity_assessment(entity)
+        ai_report = run_deepseek_summary(prompt)
 
-    # Step 2: PubMed latest research
-    papers = search_pubmed_recent(entity)
+        # Risk score
+        score = score_biosecurity_risk(ai_report)
 
-    # Step 3: Fake simple scoring logic (can be replaced with ontology check)
-    score = 85 if any(word in entity.lower() for word in ["smallpox", "anthrax", "ebola"]) else 30
-    category = "High" if score >= 70 else ("Medium" if score >= 40 else "Low")
+        # PubMed citations
+        citations = pubmed_fallback_search(entity, NCBI_API_KEY, NCBI_EMAIL)
 
-    return {
-        "entity": entity,
-        "score": score,
-        "category": category,
-        "ai_report": ai_results.get("ai_report", ""),
-        "pubmed_links": papers,
-        "timestamp": datetime.utcnow().isoformat()
-    }
+        return {
+            "entity": entity,
+            "timestamp": datetime.utcnow().isoformat(),
+            "risk_score": score,
+            "report": ai_report,
+            "citations": citations
+        }
+    except Exception as e:
+        return {
+            "entity": entity,
+            "timestamp": datetime.utcnow().isoformat(),
+            "error": str(e),
+            "risk_score": 0,
+            "report": "Error running biosecurity scan.",
+            "citations": []
+        }