aios/services/aios-v1r1-wechat-outbound-worker/worker.py

#!/usr/bin/env python3
"""AIOS V1R1 Personal WeChat Outbound Worker.

This worker is the bridge between the Xiamayiti self-evolution loop and the
already-deployed V1R1 personal WeChat capability.

Design rules:
- Do not embed credentials.
- Do not access runtime/secrets directly.
- Read owner-notification outbox JSON files.
- Deliver to a runtime-configured V1R1 notification endpoint, or try local V1R1 candidates.
- Write delivery status for audit/Web UI.
"""

from __future__ import annotations

import datetime as dt
import json
import os
import pathlib
import shutil
import time
import urllib.error
import urllib.request
from typing import Any, Dict, Iterable, List, Optional

ROOT = pathlib.Path(os.environ.get("AIOS_REPO_ROOT", "/data/aios-gitops/workspace"))
BASE = pathlib.Path(os.environ.get("AIOS_RUNTIME_ROOT", "/data/aios-gitops/runtime"))
OUTBOX = pathlib.Path(os.environ.get("AIOS_V1R1_WECHAT_OUTBOX", str(BASE / "integrations" / "v1r1-personal-wechat" / "outbox")))
SENT = pathlib.Path(os.environ.get("AIOS_V1R1_WECHAT_SENT", str(BASE / "integrations" / "v1r1-personal-wechat" / "sent")))
FAILED = pathlib.Path(os.environ.get("AIOS_V1R1_WECHAT_FAILED", str(BASE / "integrations" / "v1r1-personal-wechat" / "failed")))
STATUS = pathlib.Path(os.environ.get("AIOS_V1R1_WECHAT_STATUS", str(BASE / "integrations" / "v1r1-personal-wechat" / "status")))
AUDIT = pathlib.Path(os.environ.get("AIOS_AUDIT_DIR", str(BASE / "audit")))
INTERVAL = int(os.environ.get("AIOS_V1R1_WECHAT_INTERVAL", "10"))

# Runtime-configured endpoint. This can point to the existing V1R1 personal-WeChat module.
PRIMARY_ENDPOINT = os.environ.get("AIOS_V1R1_WECHAT_OUTBOUND_URL", "")

LOCAL_CANDIDATES = [
    "http://127.0.0.1:8788/api/v1r1/personal-wechat/send",
    "http://127.0.0.1:8788/api/v1r1/notification/owner",
    "http://127.0.0.1:8789/api/owner/wechat/send",
    "http://127.0.0.1:8790/owner/notify",
]


def now() -> str:
    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def write_json(path: pathlib.Path, value: Dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    tmp.replace(path)


def audit(action: str, status: str, detail: Dict[str, Any]) -> None:
    record = {"ts": now(), "action": action, "status": status, "detail": detail}
    safe_name = record["ts"].replace(":", "-") + f"-{action}.json"
    write_json(AUDIT / safe_name, record)
    print(json.dumps(record, ensure_ascii=False), flush=True)


def normalize_message(item: Dict[str, Any]) -> Dict[str, Any]:
    title = item.get("title") or "AIOS GitOps 通报"
    content = item.get("content") or item.get("message") or json.dumps(item, ensure_ascii=False)
    priority = item.get("priority") or "normal"
    return {
        "event_id": item.get("event_id") or item.get("id") or f"evt-{int(time.time())}",
        "channel": "personal_wechat",
        "priority": priority,
        "title": title,
        "content": content,
        "source": item.get("source") or "aios-self-evolution",
        "branch": item.get("branch"),
        "from_version": item.get("from_version"),
        "to_version": item.get("to_version"),
        "rollback_target": item.get("rollback_target") or "V1R1C4",
        "web_ui_audit_url": item.get("web_ui_audit_url") or "/portal/me/#gitops-audit",
        "created_at": item.get("created_at") or now(),
    }


def endpoints() -> List[str]:
    values: List[str] = []
    if PRIMARY_ENDPOINT:
        values.append(PRIMARY_ENDPOINT)
    values.extend(LOCAL_CANDIDATES)
    # Preserve order but remove duplicates.
    result: List[str] = []
    seen = set()
    for value in values:
        if value and value not in seen:
            result.append(value)
            seen.add(value)
    return result


def post_json(url: str, payload: Dict[str, Any]) -> Dict[str, Any]:
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}, method="POST")
    with urllib.request.urlopen(req, timeout=8) as resp:
        text = resp.read().decode("utf-8", errors="replace")[:500]
        return {"ok": 200 <= resp.status < 300, "status": resp.status, "response": text, "url": url}


def deliver(message: Dict[str, Any]) -> Dict[str, Any]:
    attempts = []
    for url in endpoints():
        try:
            result = post_json(url, message)
            attempts.append(result)
            if result.get("ok"):
                return {"status": "sent", "endpoint": url, "attempts": attempts}
        except urllib.error.URLError as exc:
            attempts.append({"ok": False, "url": url, "error": str(exc)})
        except Exception as exc:  # noqa: BLE001
            attempts.append({"ok": False, "url": url, "error": str(exc)})
    return {"status": "failed", "attempts": attempts, "reason": "no_endpoint_accepted_message"}


def process_file(path: pathlib.Path) -> None:
    try:
        item = json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:  # noqa: BLE001
        failed_target = FAILED / path.name
        FAILED.mkdir(parents=True, exist_ok=True)
        shutil.move(str(path), str(failed_target))
        audit("v1r1.wechat.parse", "failed", {"path": str(path), "error": str(exc)})
        return

    message = normalize_message(item)
    result = deliver(message)
    status_record = {"ts": now(), "file": path.name, "message": message, "delivery": result}
    write_json(STATUS / (path.stem + ".status.json"), status_record)

    if result.get("status") == "sent":
        SENT.mkdir(parents=True, exist_ok=True)
        shutil.move(str(path), str(SENT / path.name))
        audit("v1r1.wechat.deliver", "sent", status_record)
    else:
        FAILED.mkdir(parents=True, exist_ok=True)
        shutil.move(str(path), str(FAILED / path.name))
        audit("v1r1.wechat.deliver", "failed", status_record)


def process_once() -> int:
    OUTBOX.mkdir(parents=True, exist_ok=True)
    count = 0
    for path in sorted(OUTBOX.glob("*.json"), key=lambda p: p.stat().st_mtime):
        process_file(path)
        count += 1
    return count


def emit_startup_notice() -> pathlib.Path:
    OUTBOX.mkdir(parents=True, exist_ok=True)
    path = OUTBOX / f"startup-{int(time.time())}.json"
    write_json(path, {
        "event_id": f"startup-{int(time.time())}",
        "type": "owner_notification",
        "channel": "personal_wechat",
        "priority": "high",
        "title": "AIOS 自我进化系统已启动",
        "content": "AIOS 虾马一体 self-evolution loop 已进入运行态。daemon、event bus、GitOps 写链路已恢复；后续将通过 V1R1 个人微信模块与 Web UI 审计同步进展。",
        "source": "aios-v1r1-wechat-outbound-worker",
        "branch": "patch/web-ui-complete-gitops-dashboard-20260706",
        "from_version": "V1R1C4",
        "to_version": "V1R1C5",
        "rollback_target": "V1R1C4",
        "web_ui_audit_url": "/portal/me/#gitops-audit",
        "created_at": now()
    })
    return path


def main() -> int:
    import argparse
    parser = argparse.ArgumentParser(description="AIOS V1R1 personal WeChat outbound worker")
    parser.add_argument("--once", action="store_true")
    parser.add_argument("--emit-startup-notice", action="store_true")
    args = parser.parse_args()

    OUTBOX.mkdir(parents=True, exist_ok=True)
    SENT.mkdir(parents=True, exist_ok=True)
    FAILED.mkdir(parents=True, exist_ok=True)
    STATUS.mkdir(parents=True, exist_ok=True)
    AUDIT.mkdir(parents=True, exist_ok=True)

    if args.emit_startup_notice:
        path = emit_startup_notice()
        audit("v1r1.wechat.outbox", "created", {"path": str(path)})

    if args.once:
        process_once()
        return 0

    audit("v1r1.wechat.worker.start", "ok", {"outbox": str(OUTBOX), "has_primary_endpoint": bool(PRIMARY_ENDPOINT)})
    while True:
        process_once()
        time.sleep(INTERVAL)


if __name__ == "__main__":
    raise SystemExit(main())