如何用Python（BeeWare）开发Android Service实现后台TCP消息接收？

问题描述 #

我开发了一款类似PushOver的简易Android应用，内置TCP服务器可接收文本消息、记录并推送通知，这部分功能已正常运行。但GUI关闭后就无法接收消息，PushOver能实现后台持续运行，推测需要开发Service。这是我的第一款Android应用，不清楚如何基于BeeWare（Toga/Briefcase）和Python实现该功能。开发环境为Kubuntu系统，已开启手机USB调试并直接在Android设备上运行代码。

原app.py代码

import toga
from toga.style import Pack
from toga.style.pack import COLUMN, ROW

import threading, socket


Debug = True

# ----- Color codes ----------------------

RED        = '\033[31m'
GREEN      = '\033[32m'
YELLOW     = '\033[33m'
MAGENTA    = '\033[35m'
CYAN       = '\033[36m'
WHITE      = '\033[37m'
EXCEPT     = '\033[38;5;147m'  # light purple
RESET      = '\033[0m'
EventColor = ['', GREEN, YELLOW, RED]


class TCPServer(threading.Thread):
    
  def __init__(self, app):
    super().__init__()
    self.daemon = True
    self.app = app
    self.terminated = threading.Event()
    self.start()

  def Terminate(self):
    if self.is_alive():
      self.terminated.set()
      with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as ESocket:
        try:
          ESocket.settimeout(0.2)
          ESocket.connect(('localhost', 12345))
        except: pass
      self.join()
      
  def run(self):    
    def HandleClient(Conn):
      with Conn:
        message = Conn.recv(1024).decode()
        if not message: return
        self.app.loop.call_soon_threadsafe(self.app.LogMessage, message)
        #self.app.SendNotif(message)

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as SSocket:
      try:
        SSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        SSocket.bind(('192.168.0.21', 12345))
        SSocket.listen(10)
        if Debug:
          print(GREEN + 'TCP Server is listening on port 12345...' + RESET)
        try:
          while not self.terminated.is_set():
            Conn, Addr = SSocket.accept()
            if self.terminated.is_set():
              Conn.close(); break
            HandleClient(Conn)
        except Exception as E1:
          if Debug: print(RED + f'TCP Server exception (listen): {E1}' + RESET)
        finally:
          if Debug: print(YELLOW + 'TCP Server has stopped listening.' + RESET)
      except Exception as E2:
        if Debug: print(RED + f'TCP Server exception (setup): {E2}' + RESET)
        

class Application(toga.App):
    
  def startup(self):
    self.on_exit = self.exit_handler  
    self.main_window = toga.MainWindow(title=self.formal_name)
    self.log = toga.MultilineTextInput(readonly=True, style=Pack(flex=1))
    main_box = toga.Box(children=[self.log], style=Pack(direction=COLUMN, padding=10))
    self.main_window.content = main_box
    self.main_window.show()
    self.tcp_server = TCPServer(self)

  def exit_handler(self, app):
    try:  
      self.tcp_server.Terminate()
    finally:  
      return True

  def LogMessage(self, message):
    self.log.value += message + '\n'

  def SendNotif(self, message):
    self.main_window.info_dialog('New Message', message)


def main():
    return Application()

解决方案 #

要实现GUI关闭后TCP服务器持续运行，需要解耦服务器与GUI逻辑，同时配置Android后台服务（Service）并设置为前台服务（Android 8.0+强制要求后台服务显示通知）。

步骤1：调整TCP服务器与GUI解耦 #

修改app.py，让服务器不依赖GUI实例，改用文件存储消息，同时支持系统通知：

import toga
from toga.style import Pack
from toga.style.pack import COLUMN, ROW
import threading, socket, os
from datetime import datetime

Debug = True

# ----- Color codes ----------------------
RED        = '\033[31m'
GREEN      = '\033[32m'
YELLOW     = '\033[33m'
RESET      = '\033[0m'

# Android应用私有目录存储消息日志
MESSAGE_LOG_PATH = os.path.join(os.environ.get('HOME', '.'), 'message_log.txt')

class TCPServer(threading.Thread):
    def __init__(self, on_message_received=None):
        super().__init__()
        self.daemon = False  # 后台服务不能用守护线程
        self.terminated = threading.Event()
        self.on_message_received = on_message_received
        self.start()

    def Terminate(self):
        if self.is_alive():
            self.terminated.set()
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as ESocket:
                try:
                    ESocket.settimeout(0.2)
                    ESocket.connect(('localhost', 12345))
                except:
                    pass
            self.join()

    def run(self):
        def HandleClient(Conn):
            with Conn:
                message = Conn.recv(1024).decode()
                if not message:
                    return
                # 记录消息到文件
                timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                log_entry = f"[{timestamp}] {message}\n"
                with open(MESSAGE_LOG_PATH, 'a') as f:
                    f.write(log_entry)
                # 触发消息回调（通知或GUI更新）
                if self.on_message_received:
                    self.on_message_received(message)

        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as SSocket:
            try:
                SSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                SSocket.bind(('0.0.0.0', 12345))  # 绑定所有网卡
                SSocket.listen(10)
                if Debug:
                    print(GREEN + 'TCP Server is listening on port 12345...' + RESET)
                try:
                    while not self.terminated.is_set():
                        Conn, Addr = SSocket.accept()
                        if self.terminated.is_set():
                            Conn.close()
                            break
                        HandleClient(Conn)
                except Exception as E1:
                    if Debug:
                        print(RED + f'TCP Server exception (listen): {E1}' + RESET)
                finally:
                    if Debug:
                        print(YELLOW + 'TCP Server has stopped listening.' + RESET)
            except Exception as E2:
                if Debug:
                    print(RED + f'TCP Server exception (setup): {E2}' + RESET)

class Application(toga.App):
    def startup(self):
        self.on_exit = self.exit_handler
        self.main_window = toga.MainWindow(title=self.formal_name)
        self.log = toga.MultilineTextInput(readonly=True, style=Pack(flex=1))
        main_box = toga.Box(children=[self.log], style=Pack(direction=COLUMN, padding=10))
        self.main_window.content = main_box
        self.main_window.show()

        # 加载已存储的消息日志
        self.load_message_log()

        # 启动TCP服务器
        self.tcp_server = TCPServer(on_message_received=self.on_new_message)

        # 启动Android后台服务
        self.start_android_service()

    def exit_handler(self, app):
        # GUI关闭时不停止TCP服务器，留后台运行
        return True

    def load_message_log(self):
        if os.path.exists(MESSAGE_LOG_PATH):
            with open(MESSAGE_LOG_PATH, 'r') as f:
                self.log.value = f.read()

    def on_new_message(self, message):
        # 更新GUI日志
        self.log.value += f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {message}\n"
        # 发送系统通知
        self.send_system_notification(message)

    def send_system_notification(self, message):
        self.notification_service.notify(
            title="新消息",
            message=message,
            icon=None  # 可替换为应用图标路径
        )

    def start_android_service(self):
        # 仅在Android平台启动后台服务
        if self.platform.name == 'android':
            from android import AndroidActivity
            activity = AndroidActivity.getInstance()
            intent = activity.getIntent()
            intent.setComponent(activity.getPackageName(), ".PythonService")
            activity.startService(intent)

def main():
    return Application()

步骤2：配置Android后台服务 #

1. 生成Android项目：运行briefcase create android
2. 修改android/app/src/main/AndroidManifest.xml，在<application>标签内添加Service声明：

<service
    android:name=".PythonService"
    android:enabled="true"
    android:exported="false"
    android:foregroundServiceType="connectedDevice|dataSync">
</service>

3. 在android/app/src/main/java/com/yourdomain/yourapp/（替换为你的应用包名）创建PythonService.java：

package com.yourdomain.yourapp;

import android.app.Notification;
import android.app.NotificationChannel;
import android.app.NotificationManager;
import android.app.Service;
import android.content.Intent;
import android.os.Build;
import android.os.IBinder;
import androidx.core.app.NotificationCompat;

public class PythonService extends Service {
    private static final String CHANNEL_ID = "TCP_SERVER_CHANNEL";
    private static final int NOTIFICATION_ID = 1;

    @Override
    public void onCreate() {
        super.onCreate();
        createNotificationChannel();
        startForeground(NOTIFICATION_ID, buildNotification());
        startPythonServer();
    }

    private void createNotificationChannel() {
        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
            NotificationChannel channel = new NotificationChannel(
                    CHANNEL_ID,
                    "TCP服务器",
                    NotificationManager.IMPORTANCE_LOW
            );
            channel.setDescription("后台运行TCP服务器接收消息");
            NotificationManager manager = getSystemService(NotificationManager.class);
            manager.createNotificationChannel(channel);
        }
    }

    private Notification buildNotification() {
        return new NotificationCompat.Builder(this, CHANNEL_ID)
                .setContentTitle("消息服务运行中")
                .setContentText("等待接收消息")
                .setSmallIcon(R.drawable.ic_notification)
                .setPriority(NotificationCompat.PRIORITY_LOW)
                .build();
    }

    private void startPythonServer() {
        new Thread(() -> {
            try {
                PythonLoader.loadPython(this);
                PythonLoader.exec("from app import TCPServer; TCPServer(on_message_received=lambda msg: print('Received:', msg))");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }

    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }

    @Override
    public void onDestroy() {
        super.onDestroy();
        PythonLoader.exec("from app import TCPServer; TCPServer.Terminate()");
    }
}

4. 添加通知图标：将ic_notification.png放入android/app/src/main/res/drawable/目录

步骤3：打包测试 #

运行briefcase build android和briefcase run android部署到设备，关闭GUI后，用另一设备向手机IP的12345端口发送TCP消息，即可收到系统通知。

内容的提问来源于stack exchange，提问作者Marus Gradinaru