import os
import re
import json
import time
import base64
import pickle
import subprocess
import threading
import traceback
import html
import binascii
from datetime import datetime
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from flask import Flask, request, jsonify, send_from_directory
from flask_cors import CORS
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from dotenv import load_dotenv
load_dotenv()
class LogBuffer:
def __init__(self, max_items: int = 10000):
self._buf: List[Dict[str, Any]] = []
self._lock = threading.Lock()
self._next_id = 1
self._max = max_items
def append(self, msg: str, level: str = "info", source: str = "server"):
ts = datetime.now().strftime("%H:%M:%S")
line = {"id": self._next_id, "ts": ts, "level": level, "source": source, "msg": msg}
with self._lock:
self._buf.append(line)
self._next_id += 1
if len(self._buf) > self._max:
self._buf = self._buf[-self._max:]
def clear(self):
with self._lock:
self._buf.clear()
def get_after(self, after_id: int, limit: int = 500):
with self._lock:
if after_id <= 0:
data = self._buf[-limit:]
else:
data = [x for x in self._buf if x["id"] > after_id][:limit]
last_id = self._buf[-1]["id"] if self._buf else after_id
return data, last_id
logs = LogBuffer()
def log(msg: str, level: str = "info", source: str = "server"):
logs.append(msg, level, source)
print(f"[{level.upper()}][{source}] {msg}", flush=True)
def decode_base64_with_padding(b64_string: str) -> bytes:
missing_padding = len(b64_string) % 4
if missing_padding:
b64_string += "=" * (4 - missing_padding)
try:
return base64.b64decode(b64_string)
except binascii.Error as e:
log(f"Error decoding base64 string: {e}", "error", "SERVER")
return b""
WRITABLE_DIR = "/tmp"
COOKIES_PATH = os.path.join(WRITABLE_DIR, "facebook_cookies.pkl")
SERVICE_ACCOUNT_FILE = os.path.join(WRITABLE_DIR, "service_account.json")
if "FB_COOKIES_B64" in os.environ:
decoded_cookies = decode_base64_with_padding(os.environ["FB_COOKIES_B64"])
if decoded_cookies:
with open(COOKIES_PATH, "wb") as f:
f.write(decoded_cookies)
if "SERVICE_ACCOUNT_B64" in os.environ:
decoded_service_account = decode_base64_with_padding(os.environ["SERVICE_ACCOUNT_B64"])
if decoded_service_account:
with open(SERVICE_ACCOUNT_FILE, "w") as f:
f.write(decoded_service_account.decode("utf-8"))
GROUPS_TXT = os.environ.get("GROUPS_TXT", "groups.txt")
FINAL5_PATH = os.environ.get("FINAL5_PATH", "final5.py")
PYTHON_BIN = os.environ.get("PYTHON_BIN", "python")
SENDER_EMAIL = os.environ.get("SENDER_EMAIL", "smahato@hillsidemedicalgroup.com")
SCRAPE_OUTDIR = os.path.join(WRITABLE_DIR, "scraped")
ANALYSIS_OUTDIR = os.path.join(WRITABLE_DIR, "analysis")
GEMINI_KEYS = []
for i in range(1, 6):
key = os.environ.get(f"GEMINI_API_KEY_{i}")
if key:
GEMINI_KEYS.append(key)
GMAIL_SCOPES = ["https://www.googleapis.com/auth/gmail.send"]
os.makedirs(SCRAPE_OUTDIR, exist_ok=True)
os.makedirs(ANALYSIS_OUTDIR, exist_ok=True)
def build_gmail_service():
if not os.path.exists(SERVICE_ACCOUNT_FILE):
log("Service account file not found, Gmail unavailable.", "error", "GMAIL")
return None
try:
creds = service_account.Credentials.from_service_account_file(
SERVICE_ACCOUNT_FILE, scopes=GMAIL_SCOPES
).with_subject(SENDER_EMAIL)
service = build("gmail", "v1", credentials=creds)
log("Gmail service built successfully using service account.", "info", "GMAIL")
return service
except Exception as e:
log(f"Failed to build Gmail service: {e}", "error", "GMAIL")
log(f"CRITICAL: Ensure your service account has Domain-Wide Delegation enabled for the user {SENDER_EMAIL}", "error", "GMAIL")
return None
gmail_service = build_gmail_service()
@dataclass
class GroupRun:
link: str
stage: str = "pending"
scraped_json: str = ""
analysis_json: str = ""
scraped_posts: int = 0
detected_posts: int = 0
emails_sent_by_final5: int = 0
error: str = ""
@dataclass
class PipelineState:
running: bool = False
message: str = "idle"
progress: int = 0
current: int = 0
total: int = 0
groups: List[GroupRun] = field(default_factory=list)
recipients: List[str] = field(default_factory=list)
summary_path: str = ""
app = Flask(__name__, static_folder=".", static_url_path="")
CORS(app)
live_lock = threading.Lock()
live_state: Dict[str, Any] = {
"group": None,
"counts": {"total_posts": 0, "kw_hits": 0, "ai_done": 0, "confirmed": 0, "emails": 0},
"posts": []
}
def reset_live_state(group_link: str):
with live_lock:
live_state["group"] = group_link
live_state["counts"] = {"total_posts": 0, "kw_hits": 0, "ai_done": 0, "confirmed": 0, "emails": 0}
live_state["posts"] = []
def ensure_post_obj(pid: int) -> Dict[str, Any]:
with live_lock:
for p in live_state["posts"]:
if p.get("id") == pid:
return p
p = {"id": pid, "text": "", "group_link": live_state.get("group")}
live_state["posts"].append(p)
return p
def load_scraped_into_live(path: str):
try:
with open(path, "r", encoding="utf-8") as f:
posts = json.load(f)
except Exception as e:
log(f"live load error: {e}", "error", "LIVE")
return
with live_lock:
live_state["posts"] = posts
live_state["counts"]["total_posts"] = len(posts)
def handle_event_line(line: str):
if not line.startswith("::"):
return
try:
if "::SCRAPE_SAVED::" in line:
path = line.split("::SCRAPE_SAVED::", 1)[1].strip()
if path:
load_scraped_into_live(path)
elif "::KW_HIT::" in line:
d = json.loads(line.split("::KW_HIT::", 1)[1].strip())
p = ensure_post_obj(int(d["id"]))
p["found_keywords"] = d.get("found_keywords", [])
with live_lock:
live_state["counts"]["kw_hits"] += 1
elif "::AI_RESULT::" in line:
d = json.loads(line.split("::AI_RESULT::", 1)[1].strip())
p = ensure_post_obj(int(d["id"]))
ai = d.get("ai", {})
p["ai"] = ai
with live_lock:
live_state["counts"]["ai_done"] += 1
if ai.get("is_medical_seeking"):
live_state["counts"]["confirmed"] += 1
elif "::EMAIL_SENT::" in line:
d = json.loads(line.split("::EMAIL_SENT::", 1)[1].strip())
p = ensure_post_obj(int(d["id"]))
sent = int(d.get("sent", 0))
p["email_sent"] = sent > 0
if sent > 0:
with live_lock:
live_state["counts"]["emails"] += sent
except Exception as e:
log(f"live parse error: {e}", "error", "LIVE")
def read_groups(path: str) -> List[str]:
if not os.path.exists(path):
return []
with open(path, "r", encoding="utf-8") as f:
return [ln.strip() for ln in f.read().splitlines() if ln.strip()]
def slugify(url: str) -> str:
s = re.sub(r"[^a-zA-Z0-9]+", "-", url)
return s.strip("-").lower()
def send_html_email(to_emails: List[str], subject: str, html_content: str) -> int:
if not gmail_service:
log("Gmail not configured; skipping email", "warn", "gmail")
return 0
from email.message import EmailMessage
sent = 0
for to in to_emails:
try:
msg = EmailMessage()
msg["to"] = to
msg["from"] = SENDER_EMAIL
msg["subject"] = subject
msg.set_content(html_content, subtype="html")
raw = base64.urlsafe_b64encode(msg.as_bytes()).decode("utf-8")
gmail_service.users().messages().send(userId="me", body={"raw": raw}).execute()
sent += 1
log(f"Successfully sent email to {to}", "info", "GMAIL")
except HttpError as e:
log(f"Gmail HTTP error to {to}: {e}", "error", "gmail")
except Exception as e:
log(f"Gmail send error to {to}: {e}", "error", "gmail")
return sent
def build_confirmed_posts_email(groups_run: List["GroupRun"], all_confirmed_posts: List[Dict[str, Any]]) -> str:
total_groups = len(groups_run)
total_scraped = sum(g.scraped_posts for g in groups_run)
total_confirmed = len(all_confirmed_posts)
rows = []
for g in groups_run:
rows.append(f"""
| {g.link} |
{g.scraped_posts} |
{g.detected_posts} |
{"OK" if g.stage == "done" else "ERROR"} |
""")
summary_table_html = f"""Group Summary
| Group Link |
Posts Scraped |
Confirmed Posts |
Status |
{''.join(rows)}
"""
if all_confirmed_posts:
posts_html = "".join(
f"""
Post ID: {p.get("id", "N/A")} | Urgency: {p.get("ai_analysis", {}).get("urgency_level", "N/A")} | Confidence: {p.get("ai_analysis", {}).get("confidence", "N/A")}
Summary: {html.escape(p.get("ai_analysis", {}).get("medical_summary", "N/A"))}
Text:
{html.escape(p.get("text", "N/A"))}
View Group
"""
for p in all_confirmed_posts
)
else:
posts_html = "No confirmed medical posts were found during this run.
"
return f"""
Hillside Medical Group - Confirmed Medical Posts Summary
Hillside Medical Group - Confirmed Medical Posts
Run completed: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
Overall Summary: Processed {total_groups} groups, scraped {total_scraped} posts, found {total_confirmed} confirmed medical posts.
{summary_table_html}
Confirmed Posts Details
{posts_html}
This email contains posts identified as potentially seeking personal medical help. Please review and take appropriate action.
Note: The link provided is to the group. Direct post links are not currently extracted.
"""
state = PipelineState()
def stream_process_lines(args: List[str], env: Optional[Dict[str, str]] = None, tag: str = "FINAL5") -> int:
log(f"Exec: {' '.join(args)}", "info", tag)
proc = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
universal_newlines=True,
env=env or os.environ.copy()
)
def pump(pipe, name):
for raw in pipe:
line = (raw or "").rstrip("\n")
if not line:
continue
if line.startswith("::"):
try:
handle_event_line(line)
except Exception as e:
log(f"event parse error: {e}", "error", tag)
log(line, "info" if name == "stdout" else "warn", tag)
t1 = threading.Thread(target=pump, args=(proc.stdout, "stdout"), daemon=True)
t2 = threading.Thread(target=pump, args=(proc.stderr, "stderr"), daemon=True)
t1.start()
t2.start()
rc = proc.wait()
t1.join(timeout=0.2)
t2.join(timeout=0.2)
log(f"Exit code: {rc}", "info", tag)
return rc
def call_final5_for_group(group_url: str, out_json: str, analysis_json: str, recipients: List[str]) -> Dict[str, Any]:
args = [
PYTHON_BIN, FINAL5_PATH,
"--group", group_url,
"--out", out_json,
"--analysis-out", analysis_json,
"--recipients", ",".join(recipients),
"--sender", SENDER_EMAIL,
"--cookies-file", COOKIES_PATH,
"--headless"
]
if GEMINI_KEYS:
args.extend(["--gemini-keys", ",".join(GEMINI_KEYS)])
env = os.environ.copy()
env["PYTHONUNBUFFERED"] = "1"
env["PYTHONIOENCODING"] = "utf-8"
env.setdefault("HOME", WRITABLE_DIR)
env.setdefault("WDM_LOCAL", "1")
env.setdefault("WDM_CACHE_DIR", os.path.join(WRITABLE_DIR, ".wdm"))
env.setdefault("SE_MANAGER_DRIVER_CACHE", os.path.join(WRITABLE_DIR, "selenium"))
os.makedirs(env["WDM_CACHE_DIR"], exist_ok=True)
os.makedirs(env["SE_MANAGER_DRIVER_CACHE"], exist_ok=True)
rc = stream_process_lines(args, env=env, tag="FINAL5")
return {"ok": rc == 0, "code": rc}
def run_pipeline(recipients: List[str]):
try:
logs.clear()
log("Pipeline starting", "info", "ORCHESTRATOR")
state.running = True
state.message = "initializing"
state.progress = 0
state.recipients = recipients
state.groups.clear()
links = read_groups(GROUPS_TXT)
state.total = len(links)
if not links:
log("No groups found in groups.txt", "warn", "ORCHESTRATOR")
state.message = "No groups"
state.running = False
return
all_confirmed_posts = []
for i, link in enumerate(links, start=1):
reset_live_state(link)
g = GroupRun(link=link, stage="running")
state.groups.append(g)
state.current = i
state.message = f"Processing {link}"
state.progress = int(((i - 1) / max(1, state.total)) * 100)
log(f"[{i}/{state.total}] Processing group: {link}", "info", "ORCHESTRATOR")
slug = slugify(link)
out_json = os.path.join(SCRAPE_OUTDIR, f"{slug}.json")
analysis_json = os.path.join(ANALYSIS_OUTDIR, f"analysis_{slug}.json")
g.scraped_json = out_json
g.analysis_json = analysis_json
result = call_final5_for_group(link, out_json, analysis_json, recipients)
if not result.get("ok"):
g.stage = "error"
g.error = f"final5 exit code {result.get('code')}"
log(f"final5 failed for {link}: code {result.get('code')}", "error", "ORCHESTRATOR")
else:
try:
if os.path.exists(out_json):
with open(out_json, "r", encoding="utf-8") as f:
g.scraped_posts = len(json.load(f))
if os.path.exists(analysis_json):
with open(analysis_json, "r", encoding="utf-8") as f:
a = json.load(f)
g.detected_posts = a.get("confirmed_medical", 0)
g.emails_sent_by_final5 = a.get("emails_sent", 0)
confirmed_posts = a.get("posts", [])
for post in confirmed_posts:
if "group_link" not in post:
post["group_link"] = link
all_confirmed_posts.extend(confirmed_posts)
g.stage = "done"
log(f"Group done: scraped={g.scraped_posts}, confirmed={g.detected_posts}", "info", "ORCHESTRATOR")
except Exception as e:
g.stage = "error"
g.error = f"parse_error: {e}"
log(f"Parsing outputs failed for {link}: {e}", "error", "ORCHESTRATOR")
state.progress = int((i / max(1, state.total)) * 100)
try:
html_content = build_confirmed_posts_email(state.groups, all_confirmed_posts)
subject = f"🩺 Hillside - Confirmed Medical Posts Found ({len(all_confirmed_posts)} total)"
sent_count = send_html_email(recipients, subject, html_content)
log(f"Consolidated email sent to {len(recipients)} recipient(s), {sent_count} successful", "info", "GMAIL")
except Exception as e:
log(f"Error building or sending consolidated email: {e}", "error", "ORCHESTRATOR")
summary = {"run_date": datetime.now().isoformat(), "groups": [g.__dict__ for g in state.groups]}
summary_path = os.path.join(ANALYSIS_OUTDIR, "analysis_summary.json")
with open(summary_path, "w", encoding="utf-8") as f:
json.dump(summary, f, ensure_ascii=False, indent=2)
state.summary_path = summary_path
state.message = "All groups processed"
state.progress = 100
state.running = False
log("Pipeline finished", "info", "ORCHESTRATOR")
except Exception as e:
state.message = f"pipeline_error: {e}"
state.running = False
log(f"Pipeline error: {e}\n{traceback.format_exc()}", "error", "ORCHESTRATOR")
@app.route("/")
def index():
return send_from_directory(".", "index.html")
@app.get("/api/system/status")
def system_status():
return jsonify({
"gmail": gmail_service is not None,
"groups_file_exists": os.path.exists(GROUPS_TXT),
"groups_count": len(read_groups(GROUPS_TXT)),
"scrape_outdir": SCRAPE_OUTDIR,
"analysis_outdir": ANALYSIS_OUTDIR,
"sender_email": SENDER_EMAIL,
"final5_exists": os.path.exists(FINAL5_PATH),
"gemini_keys_count": len(GEMINI_KEYS)
})
@app.get("/api/groups")
def api_groups():
return jsonify({"groups": read_groups(GROUPS_TXT)})
@app.post("/api/process/start")
def api_process_start():
if state.running:
return jsonify({"success": False, "message": "Already running"}), 409
data = request.json or {}
recips = data.get("recipients") or [SENDER_EMAIL]
if isinstance(recips, str):
recips = [e.strip() for e in recips.split(",") if e.strip()]
threading.Thread(target=run_pipeline, args=(recips,), daemon=True).start()
log(f"Start requested by client; recipients={recips}", "info", "API")
return jsonify({"success": True, "message": "Pipeline started", "recipients": recips})
@app.get("/api/process/status")
def api_process_status():
return jsonify({
"running": state.running,
"message": state.message,
"progress": state.progress,
"current": state.current,
"total": state.total,
"groups": [g.__dict__ for g in state.groups]
})
@app.get("/api/process/logs")
def api_process_logs():
data, last_id = logs.get_after(int(request.args.get("after", "0")), limit=int(request.args.get("limit", "500")))
return jsonify({"entries": data, "last": last_id})
@app.post("/api/process/clear-logs")
def api_clear_logs():
logs.clear()
log("Logs cleared by client", "info", "API")
return jsonify({"success": True})
@app.get("/api/live/state")
def api_live_state():
with live_lock:
return jsonify({"success": True, "data": live_state})
@app.get("/api/results/summary")
def api_results_summary():
p = state.summary_path or os.path.join(ANALYSIS_OUTDIR, "analysis_summary.json")
if not os.path.exists(p):
return jsonify({"success": False, "message": "No summary yet"}), 404
with open(p, "r", encoding="utf-8") as f:
return jsonify({"success": True, "data": json.load(f)})
@app.get("/api/recipients")
def api_get_recipients():
recipients_path = "recipients.json"
if not os.path.exists(recipients_path):
return jsonify({"success": False, "message": "recipients.json not found"}), 404
try:
with open(recipients_path, "r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, list):
return jsonify({"success": False, "message": "Invalid format"}), 500
return jsonify({"success": True, "data": data})
except Exception as e:
return jsonify({"success": False, "message": f"Error reading file: {str(e)}"}), 500
if __name__ == "__main__":
port = int(os.environ.get("PORT", 7860))
app.run(host="0.0.0.0", port=port)