宜博 / pixelphone

#!/usr/bin/env python3
"""
App Connector abstraction layer — Sprint 8.
Each Connector class wraps one app's data access and action methods,
reading its configuration from the Knowledge Base (~/.hermes/connectors/).

Execution priority chain (per-operation):
  1. System API / ContentProvider
  2. Intent launch
  3. Frida hook
  4. Accessibility automation
"""
from __future__ import annotations

import json
import os
import sqlite3
import subprocess
import sys
import time
from pathlib import Path
from typing import Any

_TERMUX_HOME = Path("/data/data/com.termux/files/home")
_HERMES_BASE = (_TERMUX_HOME if _TERMUX_HOME.exists()
                else Path(os.environ.get("HOME", str(_TERMUX_HOME)))) / ".hermes"

CONNECTORS_DIR = Path(os.environ.get(
    "CONNECTORS_DIR",
    str(_HERMES_BASE / "connectors")
))
EVENT_BUS_DIR = Path(os.environ.get(
    "EVENT_BUS_DIR",
    str(_HERMES_BASE / "event_bus")
))
KNOWLEDGE_LOG = Path(os.environ.get(
    "KNOWLEDGE_LOG",
    str(_HERMES_BASE / "knowledge_log.jsonl")
))


# ──── Knowledge Base I/O ──────────────────────────────────────────────────────

def _load_kb(app: str) -> dict:
    path = CONNECTORS_DIR / f"{app}.yaml"
    if not path.exists():
        return {}
    try:
        import yaml
        return yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except ImportError:
        import json as _json
        return _json.loads(path.read_text(encoding="utf-8"))


def _save_kb(app: str, data: dict):
    CONNECTORS_DIR.mkdir(parents=True, exist_ok=True)
    path = CONNECTORS_DIR / f"{app}.yaml"
    try:
        import yaml
        path.write_text(yaml.dump(data, allow_unicode=True, default_flow_style=False),
                        encoding="utf-8")
    except ImportError:
        path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def _log_result(app: str, operation: str, method: str, success: bool, detail: str = ""):
    entry = {
        "ts": int(time.time()),
        "app": app,
        "operation": operation,
        "method": method,
        "success": success,
        "detail": detail,
    }
    with open(KNOWLEDGE_LOG, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    # Update KB with success/failure stats
    kb = _load_kb(app)
    stats = kb.setdefault("stats", {})
    op_stats = stats.setdefault(operation, {}).setdefault(method, {"success": 0, "fail": 0})
    op_stats["success" if success else "fail"] += 1
    op_stats["last_used"] = int(time.time())
    _save_kb(app, kb)


# ──── Base Connector ──────────────────────────────────────────────────────────

class BaseConnector:
    APP = ""

    def __init__(self):
        self._kb = _load_kb(self.APP)

    @staticmethod
    def _is_on_device() -> bool:
        return Path("/data/data/com.termux").exists()

    def _run_adb(self, cmd: str) -> str:
        if self._is_on_device():
            # Running on the device — use su -c directly
            result = subprocess.run(
                ["su", "-c", cmd],
                capture_output=True, text=True, timeout=30
            )
        else:
            result = subprocess.run(
                f"adb shell su -c {repr(cmd)}",
                shell=True, capture_output=True, text=True, timeout=30
            )
        return (result.stdout + result.stderr).strip()

    def _run_local(self, cmd: str) -> str:
        result = subprocess.run(
            cmd, shell=True, capture_output=True, text=True, timeout=30
        )
        return (result.stdout + result.stderr).strip()

    def _try_methods(self, operation: str, methods: list[tuple[str, callable]]) -> Any:
        for name, fn in methods:
            try:
                result = fn()
                _log_result(self.APP, operation, name, True)
                return result
            except Exception as e:
                _log_result(self.APP, operation, name, False, str(e))
        raise RuntimeError(f"{self.APP}.{operation}: all methods failed")


# ──── WeChat Connector ────────────────────────────────────────────────────────

class WeChatConnector(BaseConnector):
    APP = "wechat"

    PACKAGE = "com.tencent.mm"
    CIPHER_KEY = "e03ce81"

    def __init__(self):
        super().__init__()
        self._db_path = self._kb.get("db_path", "")
        if not self._db_path:
            self._db_path = self._discover_db_path()
            if self._db_path:
                kb = self._kb
                kb["db_path"] = self._db_path
                _save_kb(self.APP, kb)

    def _discover_db_path(self) -> str:
        try:
            result = self._run_adb(
                "ls /data/data/com.tencent.mm/MicroMsg/ 2>/dev/null | head -1"
            )
            for line in result.splitlines():
                if len(line.strip()) == 32:  # MD5 hash directory
                    return f"/data/data/com.tencent.mm/MicroMsg/{line.strip()}/EnMicroMsg.db"
        except Exception:
            pass
        return ""

    def _sqlcipher(self, sql: str) -> list[dict]:
        db_path = self._db_path
        if not db_path:
            return []
        # Find sqlcipher binary
        candidates = [
            "/data/data/com.termux/files/usr/bin/sqlcipher",
            "/usr/bin/sqlcipher",
        ]
        sqlcipher = next((p for p in candidates if Path(p).exists()), None)
        if not sqlcipher:
            return []
        script = (
            f'PRAGMA key="{self.CIPHER_KEY}";\n'
            f'PRAGMA cipher_compatibility=1;\n'
            f'PRAGMA cipher_page_size=1024;\n'
            f'.mode json\n'
            f'{sql.rstrip(";").strip()};\n'
        )
        try:
            result = subprocess.run(
                ["su", "-c", f'{sqlcipher} "{db_path}"'],
                input=script, capture_output=True, text=True, timeout=15
            )
            raw = result.stdout.strip()
            if not raw:
                return []
            # sqlcipher .mode json: output starts with "ok" lines then multi-line JSON array
            # Collect lines starting from the first "[" line
            lines = raw.splitlines()
            json_start = next((i for i, l in enumerate(lines) if l.strip().startswith("[")), None)
            if json_start is not None:
                json_str = "\n".join(lines[json_start:])
                parsed = json.loads(json_str)
                return parsed if isinstance(parsed, list) else [parsed]
            # Fallback: try full raw
            parsed = json.loads(raw)
            return parsed if isinstance(parsed, list) else [parsed]
        except Exception:
            return []

    def read_messages(self, contact: str = "", limit: int = 50,
                      since_ts: int = 0) -> list[dict]:
        sql = (
            "SELECT msgId, talker, content, isSend, createTime FROM message "
            "WHERE type=1 AND content NOT LIKE '%<msg>%'"
        )
        if contact:
            sql += f" AND talker LIKE '%{contact}%'"
        if since_ts:
            sql += f" AND createTime >= {since_ts * 1000}"
        sql += f" ORDER BY createTime DESC LIMIT {limit}"
        return self._sqlcipher(sql)

    def read_contacts(self, limit: int = 100) -> list[dict]:
        sql = f"SELECT username, nickname, conRemark FROM rcontact LIMIT {limit}"
        return self._sqlcipher(sql)

    def read_chatrooms(self, limit: int = 50) -> list[dict]:
        sql = f"SELECT chatroomname, memberlist FROM chatroom LIMIT {limit}"
        return self._sqlcipher(sql)

    def send_message(self, to: str, content: str) -> dict:
        def _try_accessibility():
            # Import and call send_wechat_message from android_mcp
            sys.path.insert(0, str(EVENT_BUS_DIR))
            import subprocess as _sp
            cmd = (
                f"cd /data/data/com.termux/files/home/.hermes/event_bus && "
                f"python3 -c \"import android_mcp as m; "
                f"print(m.send_wechat_message(to='{to}', content='{content}'))\""
            )
            out = self._run_adb(cmd)
            return {"method": "accessibility", "output": out}

        def _try_intent():
            escaped = content.replace("'", "\\'")
            cmd = (
                f"am start -n com.tencent.mm/.ui.LauncherUI --es username {to} && "
                f"sleep 1 && input text '{escaped}' && "
                f"input keyevent 66"
            )
            self._run_adb(cmd)
            return {"method": "intent+input", "status": "sent"}

        return self._try_methods("send_message", [
            ("frida", lambda: (_ for _ in ()).throw(NotImplementedError("frida hook not implemented"))),
            ("accessibility", _try_accessibility),
            ("intent", _try_intent),
        ])


# ──── Alipay Connector (stub) ─────────────────────────────────────────────────

class AlipayConnector(BaseConnector):
    APP = "alipay"
    PACKAGE = "com.eg.android.AlipayGphone"

    def read_bills(self, limit: int = 30) -> list[dict]:
        """Read Alipay bills. Currently stub — app not installed on test device."""
        return []

    def is_available(self) -> bool:
        result = self._run_adb(f"pm path {self.PACKAGE} 2>/dev/null")
        return "package:" in result


# ──── Mail Connector ──────────────────────────────────────────────────────────

class MailConnector(BaseConnector):
    APP = "mail"

    def read_emails(self, limit: int = 20) -> list[dict]:
        providers = [
            "content://com.android.email.provider/message",
            "content://com.google.android.gm.sapi/conversations",
        ]
        for provider in providers:
            try:
                cmd = f"content query --uri {provider} --projection subject:sender:dateSent --sort 'dateSent DESC' 2>/dev/null | head -50"
                out = self._run_adb(cmd)
                rows = []
                for line in out.splitlines():
                    if "subject=" in line or "from=" in line:
                        rows.append({"raw": line.strip()})
                if rows:
                    _log_result(self.APP, "read_emails", provider, True)
                    return rows[:limit]
            except Exception as e:
                _log_result(self.APP, "read_emails", provider, False, str(e))
        return []


# ──── Calendar Connector ──────────────────────────────────────────────────────

class CalendarConnector(BaseConnector):
    APP = "calendar"

    def read_events(self, since_ts: int = 0, limit: int = 20) -> list[dict]:
        since_ms = (since_ts or int(time.time() - 7 * 86400)) * 1000
        cmd = (
            f"content query --uri content://com.android.calendar/events "
            f"--projection title:dtstart:dtend:description "
            f"--where 'dtstart > {since_ms}' 2>/dev/null | head -50"
        )
        out = self._run_adb(cmd)
        rows = []
        for line in out.splitlines():
            if "title=" in line:
                rows.append({"raw": line.strip()})
        return rows[:limit]

    def create_event(self, title: str, start_ts: int, end_ts: int,
                      description: str = "", location: str = "") -> dict:
        cmd = (
            f"content insert --uri content://com.android.calendar/events "
            f"--bind title:s:{repr(title)} "
            f"--bind dtstart:l:{start_ts * 1000} "
            f"--bind dtend:l:{end_ts * 1000} "
            f"--bind description:s:{repr(description)} "
            f"--bind eventTimezone:s:Asia/Shanghai "
            f"--bind calendar_id:i:1 2>&1"
        )
        out = self._run_adb(cmd)
        return {"output": out, "title": title}


# ──── Execution Priority Chain ────────────────────────────────────────────────

class ExecutionChain:
    """
    Routes an operation through a prioritized list of methods.
    Falls back automatically, records success/failure to Knowledge Base.
    """
    def __init__(self, connector: BaseConnector, operation: str):
        self.connector = connector
        self.operation = operation
        self._methods: list[tuple[str, callable]] = []

    def add(self, name: str, fn: callable) -> "ExecutionChain":
        self._methods.append((name, fn))
        return self

    def run(self) -> Any:
        return self.connector._try_methods(self.operation, self._methods)


# ──── Discovery Sub-agent ─────────────────────────────────────────────────────

class DiscoveryAgent:
    """
    Explores app DB structure and hook validity, updates Knowledge Base.
    Triggered manually or when a connector method fails.
    """

    def __init__(self, app: str):
        self.app = app
        self.kb = _load_kb(app)

    def discover_wechat(self) -> dict:
        connector = WeChatConnector()
        results = {"app": "wechat", "ts": int(time.time())}

        # Discover DB path
        db_path = connector._discover_db_path()
        results["db_path"] = db_path
        if db_path:
            self.kb["db_path"] = db_path

        # Discover DB schema (table names + columns)
        if db_path:
            try:
                schema_sql = "SELECT name FROM sqlite_master WHERE type='table'"
                tables = connector._sqlcipher(schema_sql)
                results["tables"] = [r.get("name") for r in tables if r.get("name")]
            except Exception as e:
                results["tables_error"] = str(e)

        # Check WeChat version
        try:
            ver_out = connector._run_adb(
                "dumpsys package com.tencent.mm 2>/dev/null | grep versionName | head -1"
            )
            results["version"] = ver_out.strip()
            self.kb["version"] = ver_out.strip()
        except Exception:
            pass

        # Validate known hooks
        hooks = self.kb.get("hooks", {})
        results["hooks_valid"] = {}
        # (Hook validation would require frida — stub for now)
        results["hooks_valid"]["send_message"] = "unknown"

        # Save updated KB
        _save_kb(self.app, self.kb)
        results["kb_updated"] = True
        return results

    def run(self) -> dict:
        if self.app == "wechat":
            return self.discover_wechat()
        return {"app": self.app, "status": "no discovery for this app"}


# ──── Initialize Knowledge Base YAML files ────────────────────────────────────

def init_knowledge_base():
    CONNECTORS_DIR.mkdir(parents=True, exist_ok=True)

    wechat_kb = {
        "app": "wechat",
        "package": "com.tencent.mm",
        "version": "8.0.x",
        "db_path": "/data/data/com.tencent.mm/MicroMsg/653cdc2320abb089e149a3f7a7683fcf/EnMicroMsg.db",
        "cipher_key": "e03ce81",
        "tables": {
            "messages": {
                "table": "message",
                "content_field": "content",
                "sender_field": "talker",
                "time_field": "createTime",
                "type_filter": "type=1 AND content NOT LIKE '%<msg>%'",
            },
            "contacts": {
                "table": "rcontact",
                "name_field": "nickname",
                "id_field": "username",
                "remark_field": "conRemark",
            },
            "chatrooms": {
                "table": "chatroom",
                "id_field": "chatroomname",
                "members_field": "memberlist",
            },
        },
        "hooks": {
            "send_message": {
                "class": "com.tencent.mm.sdk.platformtools.ab",
                "method": "sendMessage",
                "status": "unverified",
            },
        },
        "actions": {
            "send_message": {
                "priority": ["frida", "accessibility", "intent"],
            },
            "read_messages": {
                "priority": ["sqlcipher"],
            },
        },
        "known_issues": [
            "SnsMicroMsg.db (Moments) uses different cipher key",
        ],
        "stats": {},
    }

    alipay_kb = {
        "app": "alipay",
        "package": "com.eg.android.AlipayGphone",
        "version": "unknown",
        "status": "not_installed",
        "notes": "Alipay not installed on test device. DB path TBD via Discovery.",
        "actions": {
            "read_bills": {"priority": ["sqlcipher", "scraping"]},
        },
        "stats": {},
    }

    calendar_kb = {
        "app": "calendar",
        "package": "com.google.android.calendar",
        "status": "no_account",
        "notes": "No Google account / calendar account configured on device.",
        "actions": {
            "read_events": {"priority": ["content_provider"]},
            "create_event": {"priority": ["content_provider", "intent"]},
        },
        "stats": {},
    }

    mail_kb = {
        "app": "mail",
        "notes": "No mail app installed on test device.",
        "providers": [
            "content://com.android.email.provider/message",
            "content://com.google.android.gm.sapi/conversations",
        ],
        "stats": {},
    }

    _save_kb("wechat", wechat_kb)
    _save_kb("alipay", alipay_kb)
    _save_kb("calendar", calendar_kb)
    _save_kb("mail", mail_kb)
    print(f"[connectors] Knowledge Base initialized at {CONNECTORS_DIR}")


if __name__ == "__main__":
    import sys as _sys
    if len(_sys.argv) > 1 and _sys.argv[1] == "discover":
        app = _sys.argv[2] if len(_sys.argv) > 2 else "wechat"
        agent = DiscoveryAgent(app)
        result = agent.run()
        print(json.dumps(result, ensure_ascii=False, indent=2))
    elif len(_sys.argv) > 1 and _sys.argv[1] == "init":
        init_knowledge_base()
    else:
        init_knowledge_base()
        # Test WeChatConnector
        wc = WeChatConnector()
        msgs = wc.read_messages(limit=3)
        print(f"WeChat messages: {len(msgs)}")
        contacts = wc.read_contacts(limit=3)
        print(f"WeChat contacts: {len(contacts)}")