银行账单语音播报助手

一、实际应用场景与痛点

应用场景

视障用户陈先生需要每月核对银行账单，了解自己的收支情况。传统的银行账单通常是纸质或PDF格式，包含复杂的表格、数字和术语，视障用户完全无法独立阅读。他每次都需要请家人或朋友帮忙，既麻烦又泄露隐私。一次，他不小心错过了信用卡还款，产生了不必要的利息费用。他急需一个能自动读取银行账单并语音播报收支情况的智能助手。

核心痛点

1. 账单格式复杂：PDF/Excel格式难以访问

2. 数字信息密集：金额、日期、账号等数字信息多

3. 专业术语难懂：银行术语对视障用户不友好

4. 隐私泄露风险：需他人协助查看财务信息

5. 实时性差：无法及时了解账户变动

6. 操作复杂：银行APP的无障碍支持不足

7. 错误风险：手动输入或记忆容易出错

二、核心逻辑设计

1. 输入银行账单文件（PDF/Excel/图片/短信）

2. OCR识别或直接提取文本内容

3. 智能解析账单结构

4. 提取关键信息（余额、交易记录、摘要）

5. 数据清洗和格式化

6. 生成易懂的语音播报内容

7. 语音合成播报

8. 提供问答和查询功能

9. 数据分析和趋势预测

10. 安全存储和隐私保护

三、模块化代码实现

主程序文件：bank_statement_voice_assistant.py

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

"""

银行账单语音播报助手

将银行账单转换为语音播报，帮助视障用户了解收支情况

版本：3.0.0

作者：无障碍智能助手

"""

import os

import sys

import re

import json

import csv

import sqlite3

import threading

import queue

import hashlib

from datetime import datetime, timedelta

from typing import Dict, List, Tuple, Optional, Any, Union

from dataclasses import dataclass, asdict, field

from enum import Enum

import warnings

warnings.filterwarnings('ignore')

# 图像处理和OCR

try:

import cv2

import numpy as np

import pytesseract

from PIL import Image, ImageEnhance

CV2_AVAILABLE = True

TESSERACT_AVAILABLE = True

except ImportError:

CV2_AVAILABLE = False

TESSERACT_AVAILABLE = False

print("警告: 图像处理库未安装")

# PDF处理

try:

import PyPDF2

from pdf2image import convert_from_bytes

PDF_AVAILABLE = True

except ImportError:

PDF_AVAILABLE = False

print("警告: PDF处理库未安装")

# Excel处理

try:

import pandas as pd

import openpyxl

PANDAS_AVAILABLE = True

except ImportError:

PANDAS_AVAILABLE = False

print("警告: 数据处理库未安装")

# 语音合成

try:

import pyttsx3

TTS_AVAILABLE = True

except ImportError:

TTS_AVAILABLE = False

print("警告: 语音合成库未安装")

# 音频处理

try:

import simpleaudio as sa

import wave

AUDIO_AVAILABLE = True

except ImportError:

AUDIO_AVAILABLE = False

print("警告: 音频处理库未安装")

# 数据可视化（辅助功能）

try:

import matplotlib.pyplot as plt

MATPLOTLIB_AVAILABLE = True

except ImportError:

MATPLOTLIB_AVAILABLE = False

class BankType(Enum):

"""银行类型枚举"""

ICBC = "icbc" # 工商银行

CCB = "ccb" # 建设银行

ABC = "abc" # 农业银行

BOC = "boc" # 中国银行

CMB = "cmb" # 招商银行

BCM = "bcm" # 交通银行

CEB = "ceb" # 光大银行

CIB = "cib" # 兴业银行

PINGAN = "pingan" # 平安银行

CITIC = "citic" # 中信银行

SPDB = "spdb" # 浦发银行

OTHER = "other" # 其他银行

class TransactionType(Enum):

"""交易类型枚举"""

INCOME = "income" # 收入

EXPENSE = "expense" # 支出

TRANSFER_IN = "transfer_in" # 转入

TRANSFER_OUT = "transfer_out" # 转出

WITHDRAWAL = "withdrawal" # 取现

DEPOSIT = "deposit" # 存款

FEE = "fee" # 手续费

INTEREST = "interest" # 利息

REFUND = "refund" # 退款

UNKNOWN = "unknown" # 未知

class AccountType(Enum):

"""账户类型枚举"""

SAVINGS = "savings" # 储蓄账户

CHECKING = "checking" # 支票账户

CREDIT = "credit" # 信用卡

LOAN = "loan" # 贷款账户

INVESTMENT = "investment" # 投资账户

OTHER = "other" # 其他

@dataclass

class Transaction:

"""交易记录"""

id: str

date: datetime

description: str

amount: float

currency: str

transaction_type: TransactionType

category: str

balance_after: Optional[float] = None

account_name: Optional[str] = None

account_number: Optional[str] = None

counterparty: Optional[str] = None

reference: Optional[str] = None

location: Optional[str] = None

tags: List[str] = field(default_factory=list)

def to_dict(self) -> Dict:

"""转换为字典"""

return {

'id': self.id,

'date': self.date.isoformat(),

'description': self.description,

'amount': self.amount,

'currency': self.currency,

'transaction_type': self.transaction_type.value,

'category': self.category,

'balance_after': self.balance_after,

'account_name': self.account_name,

'account_number': self.account_number,

'counterparty': self.counterparty,

'reference': self.reference,

'location': self.location,

'tags': self.tags

}

@property

def is_income(self) -> bool:

"""是否为收入"""

return self.transaction_type in [

TransactionType.INCOME,

TransactionType.TRANSFER_IN,

TransactionType.REFUND

]

@property

def is_expense(self) -> bool:

"""是否为支出"""

return self.transaction_type in [

TransactionType.EXPENSE,

TransactionType.TRANSFER_OUT,

TransactionType.WITHDRAWAL,

TransactionType.FEE

]

@dataclass

class BankAccount:

"""银行账户"""

id: str

bank_type: BankType

account_type: AccountType

account_number: str

account_name: str

balance: float

currency: str

last_updated: datetime

credit_limit: Optional[float] = None

available_credit: Optional[float] = None

due_date: Optional[datetime] = None

minimum_payment: Optional[float] = None

def to_dict(self) -> Dict:

"""转换为字典"""

return {

'id': self.id,

'bank_type': self.bank_type.value,

'account_type': self.account_type.value,

'account_number': self.account_number,

'account_name': self.account_name,

'balance': self.balance,

'currency': self.currency,

'last_updated': self.last_updated.isoformat(),

'credit_limit': self.credit_limit,

'available_credit': self.available_credit,

'due_date': self.due_date.isoformat() if self.due_date else None,

'minimum_payment': self.minimum_payment

}

@dataclass

class BankStatement:

"""银行对账单"""

id: str

account: BankAccount

period_start: datetime

period_end: datetime

transactions: List[Transaction]

opening_balance: float

closing_balance: float

total_income: float

total_expense: float

file_path: Optional[str] = None

parsed_at: datetime = field(default_factory=datetime.now)

def to_dict(self) -> Dict:

"""转换为字典"""

return {

'id': self.id,

'account': self.account.to_dict(),

'period_start': self.period_start.isoformat(),

'period_end': self.period_end.isoformat(),

'transactions': [t.to_dict() for t in self.transactions],

'opening_balance': self.opening_balance,

'closing_balance': self.closing_balance,

'total_income': self.total_income,

'total_expense': self.total_expense,

'file_path': self.file_path,

'parsed_at': self.parsed_at.isoformat()

}

def get_transactions_by_date(self, date: datetime) -> List[Transaction]:

"""获取某一天的交易记录"""

return [t for t in self.transactions

if t.date.date() == date.date()]

def get_transactions_by_type(self, trans_type: TransactionType) -> List[Transaction]:

"""按交易类型筛选"""

return [t for t in self.transactions

if t.transaction_type == trans_type]

def get_transactions_by_category(self, category: str) -> List[Transaction]:

"""按分类筛选"""

return [t for t in self.transactions

if t.category == category]

def get_top_expenses(self, limit: int = 10) -> List[Transaction]:

"""获取最大的支出"""

expenses = [t for t in self.transactions if t.is_expense]

return sorted(expenses, key=lambda x: x.amount, reverse=True)[:limit]

def get_top_incomes(self, limit: int = 10) -> List[Transaction]:

"""获取最大的收入"""

incomes = [t for t in self.transactions if t.is_income]

return sorted(incomes, key=lambda x: x.amount, reverse=True)[:limit]

class DocumentExtractor:

"""文档提取器"""

def __init__(self, config: Dict):

"""

初始化文档提取器

Args:

config: 提取器配置

"""

self.config = config

# OCR配置

self.ocr_config = config.get('ocr', {})

if TESSERACT_AVAILABLE:

# 设置Tesseract路径（如果需要）

tesseract_path = self.ocr_config.get('tesseract_path')

if tesseract_path and os.path.exists(tesseract_path):

pytesseract.pytesseract.tesseract_cmd = tesseract_path

# 缓存提取结果

self.extraction_cache = {}

def extract_text(self, filepath: str, use_cache: bool = True) -> Tuple[str, Dict]:

"""

提取文档文本

Args:

filepath: 文件路径

use_cache: 是否使用缓存

Returns:

(文本内容, 元数据)

"""

if not os.path.exists(filepath):

raise FileNotFoundError(f"文件不存在: {filepath}")

# 检查缓存

cache_key = self._get_file_hash(filepath)

if use_cache and cache_key in self.extraction_cache:

print(f"使用缓存结果: {filepath}")

return self.extraction_cache[cache_key]

ext = os.path.splitext(filepath)[1].lower()

if ext == '.txt':

text, metadata = self.extract_txt(filepath)

elif ext == '.pdf':

text, metadata = self.extract_pdf(filepath)

elif ext in ['.jpg', '.jpeg', '.png', '.bmp', '.tiff']:

text, metadata = self.extract_image(filepath)

elif ext in ['.xlsx', '.xls', '.csv']:

text, metadata = self.extract_spreadsheet(filepath)

else:

raise ValueError(f"不支持的格式: {ext}")

# 缓存结果

if use_cache:

self.extraction_cache[cache_key] = (text, metadata)

return text, metadata

def extract_txt(self, filepath: str) -> Tuple[str, Dict]:

"""提取纯文本"""

try:

with open(filepath, 'r', encoding='utf-8') as f:

content = f.read()

metadata = {

'format': 'txt',

'encoding': 'utf-8',

'size': len(content),

'chars': len(content),

'lines': content.count('\n') + 1

}

return content, metadata

except UnicodeDecodeError:

# 尝试其他编码

encodings = ['gbk', 'gb2312', 'big5', 'latin-1']

for encoding in encodings:

try:

with open(filepath, 'r', encoding=encoding) as f:

content = f.read()

metadata = {

'format': 'txt',

'encoding': encoding,

'size': len(content),

'chars': len(content),

'lines': content.count('\n') + 1

}

return content, metadata

except:

continue

raise ValueError("无法解码文本文件")

def extract_pdf(self, filepath: str) -> Tuple[str, Dict]:

"""提取PDF文本"""

if not PDF_AVAILABLE:

raise ImportError("PDF处理库未安装")

try:

text = ""

metadata = {}

with open(filepath, 'rb') as f:

pdf_reader = PyPDF2.PdfReader(f)

# 提取元数据

if pdf_reader.metadata:

metadata = {

'title': pdf_reader.metadata.get('/Title', ''),

'author': pdf_reader.metadata.get('/Author', ''),

'subject': pdf_reader.metadata.get('/Subject', ''),

'creator': pdf_reader.metadata.get('/Creator', ''),

'producer': pdf_reader.metadata.get('/Producer', ''),

'pages': len(pdf_reader.pages)

}

# 先尝试文本提取

for page_num, page in enumerate(pdf_reader.pages):

try:

page_text = page.extract_text()

if page_text and len(page_text.strip()) > 10:

text += page_text + "\n\n"

else:

# 文本太少，可能是扫描件，使用OCR

print(f"第{page_num+1}页文本过少，尝试OCR...")

page_image = self._pdf_page_to_image(f, page_num)

if page_image:

ocr_text = self._ocr_image(page_image)

text += ocr_text + "\n\n"

except Exception as e:

print(f"提取第{page_num+1}页失败: {e}")

metadata.update({

'format': 'pdf',

'chars': len(text),

'lines': text.count('\n') + 1,

'extraction_method': 'text+ocr'

})

return text, metadata

except Exception as e:

raise ValueError(f"PDF提取失败: {e}")

def extract_image(self, filepath: str) -> Tuple[str, Dict]:

"""提取图片文本"""

if not (CV2_AVAILABLE and TESSERACT_AVAILABLE):

raise ImportError("图像处理库未安装")

try:

# 读取并预处理图片

image = cv2.imread(filepath)

if image is None:

raise ValueError(f"无法读取图片: {filepath}")

# 预处理

processed = self._preprocess_image(image)

# OCR识别

text = pytesseract.image_to_string(

processed,

lang=self.ocr_config.get('language', 'chi_sim+eng'),

config=self.ocr_config.get('tesseract_config', '')

)

metadata = {

'format': os.path.splitext(filepath)[1][1:],

'width': image.shape[1],

'height': image.shape[0],

'channels': image.shape[2] if len(image.shape) > 2 else 1,

'chars': len(text),

'lines': text.count('\n') + 1

}

return text, metadata

except Exception as e:

raise ValueError(f"图片提取失败: {e}")

def extract_spreadsheet(self, filepath: str) -> Tuple[str, Dict]:

"""提取电子表格文本"""

if not PANDAS_AVAILABLE:

raise ImportError("数据处理库未安装")

try:

ext = os.path.splitext(filepath)[1].lower()

if ext == '.csv':

df = pd.read_csv(filepath, encoding='utf-8')

elif ext in ['.xlsx', '.xls']:

df = pd.read_excel(filepath)

else:

raise ValueError(f"不支持的表格格式: {ext}")

# 转换为文本

text_lines = []

# 添加表头

headers = list(df.columns)

text_lines.append("表格标题: " + ", ".join(headers))

# 添加数据

for idx, row in df.iterrows():

row_text = []

for col in df.columns:

value = row[col]

if pd.notna(value):

row_text.append(f"{col}: {value}")

if row_text:

text_lines.append(f"第{idx+1}行: " + "; ".join(row_text))

text = "\n".join(text_lines)

metadata = {

'format': ext[1:],

'rows': len(df),

'columns': len(df.columns),

'chars': len(text),

'lines': len(text_lines)

}

return text, metadata

except Exception as e:

raise ValueError(f"表格提取失败: {e}")

def _preprocess_image(self, image: np.ndarray) -> np.ndarray:

"""预处理图片提高OCR精度"""

# 转换为灰度

if len(image.shape) == 3:

gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

else:

gray = image

# 去噪

denoised = cv2.fastNlMeansDenoising(gray)

# 二值化

_, binary = cv2.threshold(

denoised,

0, 255,

cv2.THRESH_BINARY + cv2.THRESH_OTSU

)

# 形态学操作

kernel = np.ones((2, 2), np.uint8)

processed = cv2.morphologyEx(binary, cv2.MORPH_CLOSE, kernel)

return processed

def _pdf_page_to_image(self, pdf_file, page_num: int) -> Optional[Image.Image]:

"""PDF页面转换为图片"""

try:

from pdf2image import convert_from_bytes

pdf_data = pdf_file.read()

images = convert_from_bytes(pdf_data, first_page=page_num+1, last_page=page_num+1)

return images[0] if images else None

except Exception as e:

print(f"PDF转图片失败: {e}")

return None

def _ocr_image(self, image: Image.Image) -> str:

"""OCR识别图片"""

if not TESSERACT_AVAILABLE:

return ""

# 增强对比度

enhancer = ImageEnhance.Contrast(image)

enhanced = enhancer.enhance(2.0)

# OCR识别

text = pytesseract.image_to_string(

enhanced,

lang=self.ocr_config.get('language', 'chi_sim+eng'),

config=self.ocr_config.get('tesseract_config', '--psm 6')

)

return text

def _get_file_hash(self, filepath: str) -> str:

"""计算文件哈希值"""

with open(filepath, 'rb') as f:

file_hash = hashlib.md5(f.read()).hexdigest()

return file_hash

class BankStatementParser:

"""银行对账单解析器"""

def __init__(self, config: Dict):

"""

初始化解析器

Args:

config: 解析器配置

"""

self.config = config

self.bank_patterns = self.load_bank_patterns()

self.category_mapping = self.load_category_mapping()

def load_bank_patterns(self) -> Dict[BankType, Dict]:

"""加载银行识别模式"""

patterns_path = "config/bank_patterns.json"

default_patterns = self.get_default_patterns()

try:

if os.path.exists(patterns_path):

with open(patterns_path, 'r', encoding='utf-8') as f:

patterns = json.load(f)

else:

patterns = default_patterns

# 保存默认模式

os.makedirs(os.path.dirname(patterns_path), exist_ok=True)

with open(patterns_path, 'w', encoding='utf-8') as f:

json.dump(patterns, f, indent=2, ensure_ascii=False)

return patterns

except Exception as e:

print(f"加载银行模式失败，使用默认模式: {e}")

return default_patterns

def get_default_patterns(self) -> Dict:

"""获取默认银行模式"""

return {

"icbc": {

"name_keywords": ["工商银行", "ICBC"],

"account_pattern": r"账号[:：]\s*(\d{16,19})",

"balance_pattern": r"余额[:：]\s*([\d,]+\.?\d*)",

"date_pattern": r"日期[:：]\s*(\d{4}[-/]\d{1,2}[-/]\d{1,2})",

"transaction_start": r"交易明细|流水记录",

"transaction_line": r"(\d{4}[-/]\d{1,2}[-/]\d{1,2})\s+([^\d\n]+?)\s+([\d,]+\.?\d*)\s+([\d,]+\.?\d*)"

},

"ccb": {

"name_keywords": ["建设银行", "CCB"],

"account_pattern": r"账号[:：]\s*(\d{16,19})",

"balance_pattern": r"余额[:：]\s*([\d,]+\.?\d*)",

"date_pattern": r"日期[:：]\s*(\d{4}[-/]\d{1,2}[-/]\d{1,2})",

"transaction_start": r"交易明细|流水记录",

"transaction_line": r"(\d{4}[-/]\d{1,2}[-/]\d{1,2})\s+([^\d\n]+?)\s+(-?[\d,]+\.?\d*)"

},

"abc": {

"name_keywords":

如果你觉得这个工具好用，欢迎关注我！