宜博 / pixelphone

"""
Android Event Bus Server
Runs in Termux, receives events from Android APK (Accessibility, Notification, Clipboard)
and serves them to Hermes via polling.

Usage:
    uvicorn event_server:app --host 127.0.0.1 --port 8765
"""

import sqlite3
import json
import time
import os
from pathlib import Path
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from typing import Any

DB_PATH = Path(os.environ.get("EVENT_BUS_DB", str(Path.home() / ".hermes" / "event_bus" / "events.db")))
MAX_EVENTS = int(os.environ.get("EVENT_BUS_MAX", "20000"))


def get_db() -> sqlite3.Connection:
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    with get_db() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS events (
                id      INTEGER PRIMARY KEY AUTOINCREMENT,
                type    TEXT NOT NULL,
                app     TEXT,
                data    TEXT NOT NULL,
                ts      REAL NOT NULL,
                source  TEXT DEFAULT 'monitor'
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_events_type ON events(type)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_events_source ON events(source)")
        # Migration: add source column if it doesn't exist yet (for existing DBs)
        try:
            conn.execute("ALTER TABLE events ADD COLUMN source TEXT DEFAULT 'monitor'")
        except sqlite3.OperationalError:
            pass  # column already exists
        conn.commit()


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield
    # trim old events on shutdown
    _trim_events()


app = FastAPI(title="Android Event Bus", lifespan=lifespan)


class Event(BaseModel):
    type: str          # e.g. "notification", "ui_event", "clipboard", "app_lifecycle"
    app: str | None = None   # package name, e.g. "com.tencent.mm"
    data: dict[str, Any] = {}
    source: str = "monitor"  # "monitor" (dumpsys) or "apk" (Android APK)


class EventOut(BaseModel):
    id: int
    type: str
    app: str | None
    data: dict[str, Any]
    ts: float
    source: str = "monitor"


def _trim_events():
    with get_db() as conn:
        conn.execute(f"""
            DELETE FROM events WHERE id NOT IN (
                SELECT id FROM events ORDER BY id DESC LIMIT {MAX_EVENTS}
            )
        """)
        conn.commit()


@app.post("/events", status_code=201)
def post_event(event: Event):
    """Android APK posts events here."""
    with get_db() as conn:
        cursor = conn.execute(
            "INSERT INTO events (type, app, data, ts, source) VALUES (?, ?, ?, ?, ?)",
            (event.type, event.app, json.dumps(event.data), time.time(), event.source)
        )
        conn.commit()
        return {"id": cursor.lastrowid}


@app.post("/events/batch", status_code=201)
def post_events_batch(events: list[Event]):
    """Android APK posts multiple events at once."""
    if not events:
        return {"count": 0, "last_id": None}
    ts = time.time()
    rows = [(e.type, e.app, json.dumps(e.data), ts, e.source) for e in events]
    with get_db() as conn:
        conn.executemany(
            "INSERT INTO events (type, app, data, ts, source) VALUES (?, ?, ?, ?, ?)", rows
        )
        conn.commit()
        last_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    return {"count": len(events), "last_id": last_id}


@app.get("/events", response_model=list[EventOut])
def get_events(
    since_id: int = Query(default=0, description="Return events with id > since_id"),
    limit: int = Query(default=50, le=500),
    type: str | None = Query(default=None, description="Filter by event type"),
    app: str | None = Query(default=None, description="Filter by app package"),
    source: str | None = Query(default=None, description="Filter by source: 'monitor' or 'apk'"),
):
    """Hermes MCP polls here."""
    sql = "SELECT id, type, app, data, ts, source FROM events WHERE id > ?"
    params: list[Any] = [since_id]
    if type:
        sql += " AND type = ?"
        params.append(type)
    if app:
        sql += " AND app = ?"
        params.append(app)
    if source:
        sql += " AND source = ?"
        params.append(source)
    sql += " ORDER BY id ASC LIMIT ?"
    params.append(limit)

    with get_db() as conn:
        rows = conn.execute(sql, params).fetchall()

    return [
        EventOut(id=r["id"], type=r["type"], app=r["app"],
                 data=json.loads(r["data"]), ts=r["ts"],
                 source=r["source"] or "monitor")
        for r in rows
    ]


@app.get("/events/latest", response_model=list[EventOut])
def get_latest_events(
    limit: int = Query(default=20, le=200),
    type: str | None = Query(default=None),
    app: str | None = Query(default=None),
    source: str | None = Query(default=None, description="Filter by source: 'monitor' or 'apk'"),
):
    """Get the most recent N events (for debugging / initial load)."""
    sql = "SELECT id, type, app, data, ts, source FROM events WHERE 1=1"
    params: list[Any] = []
    if type:
        sql += " AND type = ?"
        params.append(type)
    if app:
        sql += " AND app = ?"
        params.append(app)
    if source:
        sql += " AND source = ?"
        params.append(source)
    sql += " ORDER BY id DESC LIMIT ?"
    params.append(limit)

    with get_db() as conn:
        rows = conn.execute(sql, params).fetchall()

    return [
        EventOut(id=r["id"], type=r["type"], app=r["app"],
                 data=json.loads(r["data"]), ts=r["ts"],
                 source=r["source"] or "monitor")
        for r in reversed(rows)
    ]


@app.get("/stats")
def get_stats():
    """Event queue stats."""
    with get_db() as conn:
        total = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0]
        last = conn.execute("SELECT id, type, app, ts FROM events ORDER BY id DESC LIMIT 1").fetchone()
        oldest = conn.execute("SELECT id, ts FROM events ORDER BY id ASC LIMIT 1").fetchone()
    return {
        "total_events": total,
        "last_event_id": last["id"] if last else None,
        "last_event_type": last["type"] if last else None,
        "last_event_app": last["app"] if last else None,
        "oldest_event_ts": oldest["ts"] if oldest else None,
        "db_path": str(DB_PATH),
        "max_events": MAX_EVENTS,
    }


@app.delete("/events")
def clear_events():
    """Clear all events (dev/debug only)."""
    with get_db() as conn:
        conn.execute("DELETE FROM events")
        conn.commit()
    return {"cleared": True}


@app.get("/health")
def health():
    return {"status": "ok", "ts": time.time()}


class XhsParams(BaseModel):
    method: str
    url: str
    headers: dict[str, str] = {}
    body: str = ""

@app.post("/xhs_params", status_code=200)
def post_xhs_params(params: XhsParams):
    """Receive forwarded XHS request from xhshook (Hook.java).
    Stored in events table as type='xhs_request' so Hermes can poll via GET /events?type=xhs_request."""
    with get_db() as conn:
        cursor = conn.execute(
            "INSERT INTO events (type, app, data, ts, source) VALUES (?, ?, ?, ?, ?)",
            ("xhs_request", "com.xingin.xhs",
             json.dumps({"method": params.method, "url": params.url,
                        "headers": params.headers, "body": params.body}),
             time.time(), "xhshook")
        )
        conn.commit()
        return {"id": cursor.lastrowid}


@app.get("/xhs_params")
def get_xhs_params(limit: int = Query(default=10, le=100)):
    """Latest XHS requests captured by xhshook. Convenience wrapper over GET /events?type=xhs_request."""
    with get_db() as conn:
        rows = conn.execute(
            "SELECT id, data, ts FROM events WHERE type='xhs_request' ORDER BY id DESC LIMIT ?",
            (limit,)
        ).fetchall()
    return [{"id": r["id"], "ts": r["ts"], **json.loads(r["data"])} for r in reversed(rows)]


# ─── XHS Proxy (Hermes → xhshook → OkHttp) ───

class XhsProxyRequest(BaseModel):
    url: str
    method: str = "GET"
    body: str = ""
    headers_extra: dict[str, str] = {}

class XhsProxyResult(BaseModel):
    proxy_id: int
    status: str          # "ok" | "error"
    http_code: int = 0
    response_body: str = ""
    error: str = ""

def _init_proxy_table():
    with get_db() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS proxy_requests (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                status TEXT DEFAULT 'pending',
                created_ts REAL,
                req_url TEXT,
                req_method TEXT,
                req_body TEXT,
                req_headers_extra TEXT,
                resp_http_code INTEGER,
                resp_body TEXT,
                resp_error TEXT
            )
        """)
        conn.commit()

_init_proxy_table()

@app.post("/xhs_proxy", status_code=201)
def submit_proxy_request(req: XhsProxyRequest):
    """Hermes submits a request to be proxied through XHS's OkHttp (with auto-signing)."""
    with get_db() as conn:
        cur = conn.execute(
            "INSERT INTO proxy_requests (status, created_ts, req_url, req_method, req_body, req_headers_extra) "
            "VALUES ('pending', ?, ?, ?, ?, ?)",
            (time.time(), req.url, req.method, req.body, json.dumps(req.headers_extra))
        )
        conn.commit()
        return {"proxy_id": cur.lastrowid}

@app.get("/xhs_proxy/pending")
def get_pending_proxy():
    """xhshook polls this. Returns oldest pending request (null JSON if none)."""
    with get_db() as conn:
        row = conn.execute(
            "SELECT id, req_url, req_method, req_body, req_headers_extra FROM proxy_requests "
            "WHERE status='pending' ORDER BY id ASC LIMIT 1"
        ).fetchone()
    if not row:
        return None
    # Mark as processing
    with get_db() as conn:
        conn.execute("UPDATE proxy_requests SET status='processing' WHERE id=?", (row["id"],))
        conn.commit()
    return {
        "proxy_id": row["id"],
        "url": row["req_url"],
        "method": row["req_method"],
        "body": row["req_body"],
        "headers_extra": json.loads(row["req_headers_extra"] or "{}")
    }

@app.post("/xhs_proxy/result", status_code=200)
def submit_proxy_result(result: XhsProxyResult):
    """xhshook posts the API response here."""
    with get_db() as conn:
        conn.execute(
            "UPDATE proxy_requests SET status=?, resp_http_code=?, resp_body=?, resp_error=? WHERE id=?",
            (result.status, result.http_code, result.response_body, result.error, result.proxy_id)
        )
        conn.commit()
    return {"ok": True}

@app.get("/xhs_proxy/result/{proxy_id}")
def get_proxy_result(proxy_id: int):
    """Hermes polls this for the result."""
    with get_db() as conn:
        row = conn.execute(
            "SELECT * FROM proxy_requests WHERE id=?", (proxy_id,)
        ).fetchone()
    if not row:
        raise HTTPException(404, "proxy_id not found")
    return {
        "proxy_id": row["id"],
        "status": row["status"],
        "created_ts": row["created_ts"],
        "http_code": row["resp_http_code"],
        "response_body": row["resp_body"],
        "error": row["resp_error"]
    }