import os import json import uuid import datetime import hashlib import secrets from pathlib import Path from fastapi import FastAPI, HTTPException, Header, Response, Request from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse, RedirectResponse from pydantic import BaseModel from groq import Groq from typing import List, Optional app = FastAPI() client = Groq(api_key=os.environ.get("GROQ_API_KEY")) # ── Persistent storage ─────────────────────────────────────────────────────── _preferred = Path("/data/pharma_chat") _fallback = Path(os.environ.get("HOME", "/home/user")) / "pharma_chat" try: _preferred.mkdir(parents=True, exist_ok=True) DATA_DIR = _preferred except PermissionError: _fallback.mkdir(parents=True, exist_ok=True) DATA_DIR = _fallback THREADS_FILE = DATA_DIR / "threads.json" SESSIONS_FILE = DATA_DIR / "sessions.json" def load_threads() -> dict: if THREADS_FILE.exists(): try: return json.loads(THREADS_FILE.read_text()) except Exception: return {} return {} def save_threads(threads: dict): THREADS_FILE.write_text(json.dumps(threads, indent=2)) def load_sessions() -> dict: if SESSIONS_FILE.exists(): try: return json.loads(SESSIONS_FILE.read_text()) except Exception: return {} return {} def save_sessions(sessions: dict): SESSIONS_FILE.write_text(json.dumps(sessions, indent=2)) # ── Auth ───────────────────────────────────────────────────────────────────── # Dummy credentials — change these for production USERS = { "admin": {"password": "pharma123", "role": "Admin", "name": "Admin User"}, "auditor": {"password": "audit2024", "role": "Auditor", "name": "Sarah Chen"}, "reviewer": {"password": "review99", "role": "Reviewer", "name": "James Patel"}, "analyst": {"password": "analyst01", "role": "Analyst", "name": "Maria Lopez"}, } SESSION_TTL_HOURS = 8 def create_session(username: str) -> str: token = secrets.token_hex(32) sessions = load_sessions() sessions[token] = { "username": username, "created_at": datetime.datetime.utcnow().isoformat(), "expires_at": (datetime.datetime.utcnow() + datetime.timedelta(hours=SESSION_TTL_HOURS)).isoformat(), } save_sessions(sessions) return token def get_session(token: str) -> Optional[dict]: if not token: return None sessions = load_sessions() s = sessions.get(token) if not s: return None if datetime.datetime.utcnow() > datetime.datetime.fromisoformat(s["expires_at"]): del sessions[token] save_sessions(sessions) return None return s def require_auth(authorization: Optional[str] = Header(default=None)): token = None if authorization and authorization.startswith("Bearer "): token = authorization[7:] s = get_session(token) if not s: raise HTTPException(status_code=401, detail="Not authenticated") return s # ── Models ─────────────────────────────────────────────────────────────────── MODELS = { "llama-3.1-8b-instant": {"name": "Llama 3.1 8B", "description": "Fast responses", "icon": "⚡"}, "llama-3.3-70b-versatile": {"name": "Llama 3.3 70B", "description": "Deep analysis", "icon": "🧠"}, } SYSTEM_PROMPT = """You are PharmaComply AI, an expert pharmaceutical compliance assistant with deep knowledge across: REGULATORY FRAMEWORKS: - FDA (21 CFR Parts 11, 210, 211, 312, 314, 820) — cGMP, IND/NDA/ANDA submissions, 510(k) - EMA guidelines, ICH Q10 Pharmaceutical Quality System, ICH E6 GCP - EU GMP Annex 1-21, EudraLex Volume 4 - WHO GMP guidelines, PIC/S standards PHARMACOVIGILANCE (PV): - ICH E2A-E2F guidelines, EMA GVP modules I-XVI - FAERS/EudraVigilance reporting, MedWatch, PSUR/PBRER preparation - Signal detection, risk management (REMS/RMP), aggregate safety reporting - Serious Adverse Event (SAE) reporting timelines (7-day, 15-day, expedited) CLINICAL TRIALS / GCP: - ICH E6(R2/R3) Good Clinical Practice - Protocol deviations, CAPA management, audit findings - Informed consent regulations, IRB/IEC requirements - eCTD format, TMF structure (DIA Reference Model) QUALITY SYSTEMS: - CAPA (Corrective and Preventive Action) workflows - Deviation management, OOS/OOT investigations - Change control, validation (CSV/computerized systems, 21 CFR Part 11) - Batch record review, release procedures REPORTING: When asked to generate a report, produce structured output with: - Document header (title, date, classification) - Executive summary - Regulatory basis / applicable guidelines - Findings with risk classification (Critical / Major / Minor / Observation) - CAPA recommendations with timelines - Conclusion and sign-off section COMPLIANCE RULES: - Always cite specific regulation section numbers (e.g., 21 CFR 211.68, ICH Q10 Section 3.2) - Flag Critical findings that require immediate action - Distinguish between regulatory requirements and best practices - Never provide legal advice; recommend qualified regulatory counsel for legal matters - Apply ALCOA+ principles throughout Maintain professional, precise language appropriate for regulatory submissions and audit documentation.""" # ── Pydantic models ────────────────────────────────────────────────────────── class LoginRequest(BaseModel): username: str password: str class Message(BaseModel): role: str content: str class ChatRequest(BaseModel): thread_id: str message: str model: str = "llama-3.1-8b-instant" class NewThreadRequest(BaseModel): title: Optional[str] = None class RenameRequest(BaseModel): title: str class ReportRequest(BaseModel): thread_id: str report_type: str model: str = "llama-3.3-70b-versatile" # ── Auth endpoints ─────────────────────────────────────────────────────────── @app.post("/api/login") def login(req: LoginRequest): user = USERS.get(req.username) if not user or user["password"] != req.password: raise HTTPException(status_code=401, detail="Invalid username or password") token = create_session(req.username) return {"ok": True, "token": token, "name": user["name"], "role": user["role"], "username": req.username} @app.post("/api/logout") def logout(authorization: Optional[str] = Header(default=None)): if authorization and authorization.startswith("Bearer "): token = authorization[7:] sessions = load_sessions() sessions.pop(token, None) save_sessions(sessions) return {"ok": True} @app.get("/api/me") def me(authorization: Optional[str] = Header(default=None)): token = None if authorization and authorization.startswith("Bearer "): token = authorization[7:] s = get_session(token) if not s: raise HTTPException(status_code=401, detail="Not authenticated") u = USERS[s["username"]] return {"username": s["username"], "name": u["name"], "role": u["role"]} # ── Thread endpoints ───────────────────────────────────────────────────────── @app.get("/api/threads") def get_threads(authorization: Optional[str] = Header(default=None)): require_auth(authorization) threads = load_threads() result = [ {"id": tid, "title": t.get("title","Untitled"), "created_at": t.get("created_at",""), "updated_at": t.get("updated_at",""), "message_count": len(t.get("messages",[]))} for tid, t in threads.items() ] result.sort(key=lambda x: x["updated_at"], reverse=True) return result @app.post("/api/threads") def create_thread(req: NewThreadRequest, authorization: Optional[str] = Header(default=None)): s = require_auth(authorization) threads = load_threads() tid = str(uuid.uuid4()) now = datetime.datetime.utcnow().isoformat() threads[tid] = {"id": tid, "title": req.title or "New Conversation", "created_at": now, "updated_at": now, "created_by": s["username"], "messages": []} save_threads(threads) return threads[tid] @app.get("/api/threads/{thread_id}") def get_thread(thread_id: str, authorization: Optional[str] = Header(default=None)): require_auth(authorization) threads = load_threads() if thread_id not in threads: raise HTTPException(status_code=404, detail="Thread not found") return threads[thread_id] @app.patch("/api/threads/{thread_id}") def rename_thread(thread_id: str, req: RenameRequest, authorization: Optional[str] = Header(default=None)): require_auth(authorization) threads = load_threads() if thread_id not in threads: raise HTTPException(status_code=404, detail="Thread not found") threads[thread_id]["title"] = req.title threads[thread_id]["updated_at"] = datetime.datetime.utcnow().isoformat() save_threads(threads) return {"ok": True} @app.delete("/api/threads/{thread_id}") def delete_thread(thread_id: str, authorization: Optional[str] = Header(default=None)): require_auth(authorization) threads = load_threads() if thread_id not in threads: raise HTTPException(status_code=404, detail="Thread not found") del threads[thread_id] save_threads(threads) return {"ok": True} # ── Chat endpoint ──────────────────────────────────────────────────────────── @app.post("/api/chat") def chat(req: ChatRequest, authorization: Optional[str] = Header(default=None)): require_auth(authorization) threads = load_threads() if req.thread_id not in threads: raise HTTPException(status_code=404, detail="Thread not found") thread = threads[req.thread_id] thread["messages"].append({"role": "user", "content": req.message}) history = [{"role": m["role"], "content": m["content"]} for m in thread["messages"]] full_response = [] def generate(): stream = client.chat.completions.create( model=req.model, messages=[{"role": "system", "content": SYSTEM_PROMPT}] + history, stream=True, max_tokens=4096, ) for chunk in stream: delta = chunk.choices[0].delta if delta.content: full_response.append(delta.content) yield f"data: {json.dumps({'content': delta.content})}\n\n" assistant_text = "".join(full_response) thread["messages"].append({"role": "assistant", "content": assistant_text}) thread["updated_at"] = datetime.datetime.utcnow().isoformat() if len(thread["messages"]) == 2 and thread["title"] == "New Conversation": thread["title"] = req.message[:50] + ("…" if len(req.message) > 50 else "") threads[req.thread_id] = thread save_threads(threads) yield "data: [DONE]\n\n" return StreamingResponse( generate(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive", } ) # ── Report endpoint ────────────────────────────────────────────────────────── @app.post("/api/report") def generate_report(req: ReportRequest, authorization: Optional[str] = Header(default=None)): require_auth(authorization) threads = load_threads() if req.thread_id not in threads: raise HTTPException(status_code=404, detail="Thread not found") thread = threads[req.thread_id] history = [{"role": m["role"], "content": m["content"]} for m in thread["messages"]] report_prompts = { "compliance_summary": "Generate a formal pharmaceutical compliance summary report based on this conversation. Include: Executive Summary, Regulatory Basis, Key Findings with risk classification (Critical/Major/Minor/Observation), CAPA Recommendations with timelines, and Conclusion. Format as a professional regulatory document.", "deviation_report": "Generate a formal deviation report based on this conversation. Include: Deviation Description, Root Cause Analysis, Immediate Containment Actions, Regulatory Impact Assessment, CAPA Plan with owners and due dates. Use GMP deviation report format.", "audit_findings": "Generate a formal audit findings report based on this conversation. Include: Audit Scope, Applicable Regulations, Findings Table (Finding #, Description, Classification, Reference, CAPA), Risk Summary, and Closing Statement.", "pv_assessment": "Generate a pharmacovigilance assessment report. Include: Safety Signal Summary, ICH E2A/E2E classification, Reporting Obligations (7-day/15-day/PSUR), Risk-Benefit Assessment, and Regulatory Action Required.", "regulatory_briefing": "Generate a regulatory affairs briefing document. Include: Regulatory Strategy Overview, Applicable Guidelines, Submission Requirements, Timeline, Open Issues, and Recommendations.", } prompt = report_prompts.get(req.report_type, report_prompts["compliance_summary"]) def generate(): stream = client.chat.completions.create( model=req.model, messages=[{"role": "system", "content": SYSTEM_PROMPT}, *history, {"role": "user", "content": prompt}], stream=True, max_tokens=4096, ) for chunk in stream: delta = chunk.choices[0].delta if delta.content: yield f"data: {json.dumps({'content': delta.content})}\n\n" yield "data: [DONE]\n\n" return StreamingResponse( generate(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive", } ) # ── Frontend ───────────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) def root(): return HTMLResponse(content=HTML) HTML = r""" PharmaComply AI
"""