Manifest.xml文件注入修改

def injectManifest(dump: GameDump):
    # 解包后的 AndroidManifest.xml 路径
    manifest_path = dump.get_android_manifest_xml_path()
    ET.register_namespace('android', 'http://schemas.android.com/apk/res/android')
    # manifest_path = r'C:\Users\shubao67\Desktop\ybtjd\AndroidManifest.xml'

    print(manifest_path)

    # # 解析 XML 文件
    tree = ET.parse(manifest_path)
    root = tree.getroot()

    manifest = dump.get_manifest()
    print('injectManifest package_name:', manifest.get_package_name())

    # 定义要添加的 <provider> 元素，替换命名空间前缀 用于感知生命周期
    provider = ET.Element('provider')
    provider.set(f'android:name', 'com.ssy185.sdk.common.base.lifecycle.GmLifecycleInitializer')
    provider.set(f'android:authorities', f'{manifest.get_package_name()}.GmLifecycleInitializer')
    provider.set(f'android:exported', 'false')
    provider.set(f'android:multiprocess', 'true')

    service = ET.Element('service')
    service.set(f'android:name', 'com.ssy185.sdk.feature.record.ScreenRecordService')
    service.set(f'android:foregroundServiceType', 'mediaProjection')
    # service.set(f'tools:targetApi', 'q')

    # < uses - permission android: name = "android.permission.SYSTEM_ALERT_WINDOW" / >
    alert_window_permission = ET.Element('uses-permission')
    alert_window_permission.set(f'android:name', 'android.permission.SYSTEM_ALERT_WINDOW')

    record_audio_permission = ET.Element('uses-permission')
    record_audio_permission.set(f'android:name', 'android.permission.RECORD_AUDIO')

    foreground_service_permission = ET.Element('uses-permission')
    foreground_service_permission.set(f'android:name', 'android.permission.FOREGROUND_SERVICE')

    foreground_service_media_projection_permission = ET.Element('uses-permission')
    foreground_service_media_projection_permission.set(f'android:name', 'android.permission.FOREGROUND_SERVICE_MEDIA_PROJECTION')


    # 查找 <application> 标签并添加 <provider> 元素
    application = root.find('application')
    if application is not None:
        application.append(provider)
        application.append(service)
        request_legacy_external_storage = application.attrib.get("{http://schemas.android.com/apk/res/android}requestLegacyExternalStorage")
        if request_legacy_external_storage is None:
            # 如果没有该属性，添加并设置为 true
            application.set('android:requestLegacyExternalStorage', 'true')
            print("属性 'android:requestLegacyExternalStorage' 已添加并设置为 true")
        else:
            print("属性 'android:requestLegacyExternalStorage' 已存在")
    root.append(alert_window_permission)
    root.append(record_audio_permission)
    root.append(foreground_service_permission)
    root.append(foreground_service_media_projection_permission)

    activities = application.findall('activity')
    # 遍历所有Activity 小窗模式需要resizeableActivity 为 true
    for activity in activities:
        resizeable_activity_attr = '{http://schemas.android.com/apk/res/android}resizeableActivity'
        if resizeable_activity_attr in activity.attrib:
            print('injectManifest modify resizeableActivity to true is success!')
            activity.set(resizeable_activity_attr, 'true')

    # 将修改后的 XML 保存回文件
    tree.write(manifest_path, encoding='utf-8', xml_declaration=True)
    print(f"Provider added successfully to AndroidManifest.xml")
if __name__ == '__main__':
    injectManifest(None)

批量ADB安装应用

def process_file(file_path):
    """
    处理每个文件的函数，执行 inject_web.main 命令
    """
    command = ["adb", "install", file_path]
    print(f"正在处理: {file_path}")
    subprocess.run(command)


def foldInject(folder_paths, max_concurrent_tasks):
    """
    遍历多个文件夹路径，批量处理 APK 文件，限制每次最多并发执行 `max_concurrent_tasks` 个任务。
    """
    # 创建一个线程池，max_concurrent_tasks 确定每次最多并发的任务数
    with ThreadPoolExecutor(max_workers=max_concurrent_tasks) as executor:
        futures = []  # 用于存储所有任务的未来对象

        for folder_path in folder_paths:
            # 确保 folder_path 是一个有效路径
            folder_path = Path(folder_path)
            if not folder_path.is_dir():
                print(f"警告: 路径 {folder_path} 不是有效的目录。跳过...")
                continue

            # 遍历目录中的文件
            for root, dirs, files in os.walk(folder_path):
                for file in files:
                    # 这里可以根据需要筛选文件类型，例如只处理 .apk 文件
                    if file.endswith("_patched_signed.apk"):
                        file_path = os.path.join(root, file)
                        # 提交每个文件的处理任务到线程池
                        futures.append(executor.submit(process_file, file_path))

        # 使用 wait 来确保每批任务完成后再继续下一批
        while futures:
            # 等待最多 `max_concurrent_tasks` 个任务完成
            done, futures = wait(futures, return_when="FIRST_COMPLETED")
if __name__ == '__main__':
    # 文件夹路径数组
    folder_paths = [
        r"E:\game\webView2\test",
        # r"E:\game\webView3"
    ]
    # 设置每批最多并发执行 5 个任务
    foldInject(folder_paths, max_concurrent_tasks=5)