# Mac
brew install android-platform-tools
# Ubuntu
sudo apt install android-tools-adb

pip install adb-shell flask ddddocr

import subprocess
import re
 
def get_sms():
    try:
        # 執(zhí)行ADB命令獲取短信
        result = subprocess.run(
            ["adb", "shell", "content", "query", "--uri", "content://sms/inbox", "--projection", "address,body,date"],
            capture_output=True,
            text=True
        )
        return result.stdout
    except Exception as e:
        print(f"獲取短信失敗: {e}")
        return None
 
def parse_sms(sms_data):
    messages = sms_data.split("\n")
    for message in messages:
        if "body=" in message:
            # 提取短信內(nèi)容
            body_start = message.index("body=") + len("body=")
            body_end = message.index(",", body_start)
            body = message[body_start:body_end]
            # 提取發(fā)送者號(hào)碼
            address_start = message.index("address=") + len("address=")
            address_end = message.index(",", address_start)
            address = message[address_start:address_end]
            yield address, body
 
# 示例：獲取并解析短信
sms_data = get_sms()
if sms_data:
    for address, body in parse_sms(sms_data):
        print(f"發(fā)送者: {address}, 內(nèi)容: {body}")

def extract_verification_code(sms_body):
    # 假設(shè)驗(yàn)證碼為6位數(shù)字
    code = re.search(r'\b\d{6}\b', sms_body)
    return code.group(0) if code else None

import ddddocr
 
def ocr_verification_code(image_path):
    ocr = ddddocr.DdddOcr()
    with open(image_path, 'rb') as f:
        image = f.read()
    return ocr.classification(image)

from flask import Flask, request
 
app = Flask(__name__)
 
@app.route('/forward', methods=['POST'])
def forward():
    data = request.json
    code = data.get('code')
    # 轉(zhuǎn)發(fā)驗(yàn)證碼到目標(biāo)地址
    # 例如：requests.post(url, json={'code': code})
    return {'status': 'success', 'message': f'驗(yàn)證碼 [code] 已轉(zhuǎn)發(fā)'}
 
if __name__ == '__main__':
    app.run(port=5000)

import threading
 
def monitor_device(device_id):
    while True:
        sms_data = get_sms(device_id)
        if sms_data:
            for address, body in parse_sms(sms_data):
                code = extract_verification_code(body)
                if code:
                    forward(code)
 
# 啟動(dòng)多個(gè)設(shè)備監(jiān)控線程
devices = ['device1', 'device2', 'device3']
for device in devices:
    thread = threading.Thread(target=monitor_device, args=(device,))
    thread.start()

import subprocess
import re
from flask import Flask, request
import threading
 
app = Flask(__name__)
 
def get_sms(device_id=None):
    # 根據(jù)設(shè)備ID獲取短信（多設(shè)備支持）
    command = ["adb", "shell", "content", "query", "--uri", "content://sms/inbox", "--projection", "address,body,date"]
    if device_id:
        command.insert(1, f"-s {device_id}")
    result = subprocess.run(command, capture_output=True, text=True)
    return result.stdout
 
def parse_sms(sms_data):
    messages = sms_data.split("\n")
    for message in messages:
        if "body=" in message:
            body_start = message.index("body=") + len("body=")
            body_end = message.index(",", body_start)
            body = message[body_start:body_end]
            address_start = message.index("address=") + len("address=")
            address_end = message.index(",", address_start)
            address = message[address_start:address_end]
            yield address, body
 
def extract_verification_code(sms_body):
    code = re.search(r'\b\d{6}\b', sms_body)
    return code.group(0) if code else None
 
@app.route('/forward', methods=['POST'])
def forward():
    data = request.json
    code = data.get('code')
    # 實(shí)現(xiàn)轉(zhuǎn)發(fā)邏輯
    print(f"驗(yàn)證碼 [code] 已轉(zhuǎn)發(fā)")
    return {'status': 'success'}
 
def monitor_device(device_id):
    while True:
        sms_data = get_sms(device_id)
        if sms_data:
            for address, body in parse_sms(sms_data):
                code = extract_verification_code(body)
                if code:
                    # 調(diào)用轉(zhuǎn)發(fā)接口
                    app.config['DEBUG'] = False
                    app.config['TESTING'] = False
                    with app.app_context():
                        forward_result = forward(json={'code': code})
                    print(forward_result)
 
if __name__ == '__main__':
    # 啟動(dòng)多個(gè)設(shè)備監(jiān)控線程
    devices = ['device1', 'device2']  # 替換為實(shí)際設(shè)備ID
    for device in devices:
        thread = threading.Thread(target=monitor_device, args=(device,))
        thread.start()
    app.run(port=5000)