Create biosecurity.py

genesis/biosecurity.py

@@ -0,0 +1,105 @@
+# genesis/biosecurity.py
+import os
+import requests
+from datetime import datetime
+from typing import Dict, Any, List
+
+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 []
+
+def ai_biosecurity_assessment(entity: str) -> Dict[str, Any]:
+    """Run AI-powered biosecurity risk assessment."""
+    import openai
+    openai.api_key = OPENAI_API_KEY
+    try:
+        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
+        """
+        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)
+
+    # Step 2: PubMed latest research
+    papers = search_pubmed_recent(entity)
+
+    # 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")
+
+    return {
+        "entity": entity,
+        "score": score,
+        "category": category,
+        "ai_report": ai_results.get("ai_report", ""),
+        "pubmed_links": papers,
+        "timestamp": datetime.utcnow().isoformat()
+    }