[Python] 纯文本查看 复制代码
import tkinter as tk
from tkinter import filedialog, messagebox, ttk, Toplevel, StringVar, Checkbutton, Button
import logging
import pdfplumber
import re
import os
import subprocess
import xlwt
import datetime
pdf_files_folder = None
# 配置日志记录
logging.basicConfig(filename='app.log',
filemode='w',
format='%(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
reverse = False
# 定义排序函数
def sorter(tree, column, data_type, reverse):
l = [(tree.set(k, column), k) for k in tree.get_children('')]
if data_type == 'num':
try:
l = [(float(x), k) for x, k in l]
except ValueError as e:
pass
l.sort(reverse=reverse)
for index, (val, k) in enumerate(l):
tree.move(k, '', index)
def column_sorter(tree, column, data_type='str'):
global reverse
reverse = not reverse
sorter(tree, column, data_type, reverse)
def is_patent_fee_receipt(text):
"""检测是否是专利年费票据（非税收入票据）"""
return ('非税收入' in text or '票据(电子)' in text) and ('专利' in text or '年费' in text)
def extract_patent_fee_data(text):
"""提取专利年费票据数据"""
result = {
"is_patent": False,
"seller_name": "",
"tax_id": "",
"amount": 0.0,
"tax_amount": 0.0,
"total_amount": 0.0,
"invoice_number": "",
"date": None,
"category": "专利年费"
}
if not is_patent_fee_receipt(text):
return result
result["is_patent"] = True
logging.info("检测到专利年费票据")
invoice_match = re.search(r'票据号码[：:]\s*(\d+)', text)
if invoice_match:
result["invoice_number"] = invoice_match.group(1)
seller_patterns = [
r'(国家知识产权局[^\n]*)',
r'国家知识产权局[^\n]*专利局',
r'(国家[^,，\n]*?专利局)',
]
for pattern in seller_patterns:
match = re.search(pattern, text)
if match:
result["seller_name"] = match.group(1).strip()
break
if not result["seller_name"]:
patent_match = re.search(r'(国家[^\n]*?专利局)', text)
if patent_match:
result["seller_name"] = patent_match.group(1).strip()
else:
result["seller_name"] = "国家知识产权局专利局"
date_match = re.search(r'开票日期[：:]\s*(\d{4}-\d{1,2}-\d{1,2})', text)
if date_match:
result["date"] = normalize_date(date_match.group(1))
else:
date_match = re.search(r'开票日期[：:]\s*(\d{4}年\d{1,2}月\d{1,2}日)', text)
if date_match:
result["date"] = normalize_date(date_match.group(1))
project_match = re.search(r'([\u4e00-\u9fa5]+(?:专利第[一二三四五六七八九十\d]+年年费|年费|专利))', text)
if project_match:
result["category"] = project_match.group(1).strip()
else:
project_match = re.search(r'([\u4e00-\u9fa5]*专利[\u4e00-\u9fa5]*年费)', text)
if project_match:
result["category"] = project_match.group(1).strip()
else:
project_match = re.search(r'([\u4e00-\u9fa5]*年费[\u4e00-\u9fa5]*)', text)
if project_match:
result["category"] = project_match.group(1).strip()
amount_match = re.search(r'(?:金额合计|合\s*计)[（(]小写[）)]\s*[￥¥]?\s*(\d+(?:\.\d{1,2})?)', text)
if amount_match:
result["amount"] = float(amount_match.group(1))
result["total_amount"] = result["amount"]
result["tax_amount"] = 0.0
else:
amount_match = re.search(r'(?:年费|专利|缴费)[^\n]*?(\d+(?:\.\d{1,2})?)\s*$', text, re.MULTILINE)
if amount_match:
result["amount"] = float(amount_match.group(1))
result["total_amount"] = result["amount"]
result["tax_amount"] = 0.0
result["tax_id"] = "非税收入无税号"
logging.info(f"专利年费-提取结果: 销售方={result['seller_name']}, 金额={result['amount']}, 项目={result['category']}")
return result
def is_railway_ticket(text):
return '铁路电子客票' in text or ('12306' in text and '铁路' in text)
def extract_railway_ticket_data(text):
result = {
"is_railway": False,
"seller_name": "",
"tax_id": "",
"buyer_name": "",
"buyer_tax_id": "",
"amount": 0.0,
"tax_amount": 0.0,
"total_amount": 0.0,
"invoice_number": "",
"date": None,
"category": "铁路客运"
}
if not is_railway_ticket(text):
return result
result["is_railway"] = True
logging.info("检测到铁路电子客票")
invoice_match = re.search(r'发票号码[：:]\s*(\d+)', text)
if invoice_match:
result["invoice_number"] = invoice_match.group(1)
date_match = re.search(r'开票日期[：:]\s*(\d{4}年\d{1,2}月\d{1,2}日)', text)
if date_match:
result["date"] = normalize_date(date_match.group(1))
fare_match = re.search(r'票价[：:]\s*[￥¥]?\s*(\d{1,4}\.\d{1,2})', text)
if not fare_match:
fare_match = re.search(r'[￥¥]\s*(\d{1,4}\.\d{2})\s*\n', text)
if fare_match:
result["amount"] = float(fare_match.group(1))
result["total_amount"] = result["amount"]
result["tax_amount"] = 0.0
else:
fare_match = re.search(r'[￥¥]\s*(\d{1,3}\.\d{2})', text)
if fare_match:
val = float(fare_match.group(1))
if val < 10000:
result["amount"] = val
result["total_amount"] = val
result["tax_amount"] = 0.0
buyer_match = re.search(r'购买方名称[：:]\s*([^\n]+)', text)
if buyer_match:
buyer_text = buyer_match.group(1).strip()
parts = re.split(r'\s+统一社会信用代码[：:]', buyer_text)
if len(parts) >= 2:
result["buyer_name"] = parts[0].strip()
result["buyer_tax_id"] = parts[1].strip()
else:
result["buyer_name"] = buyer_text
result["seller_name"] = "国家铁路局"
result["tax_id"] = "铁路客票无税号"
stations = re.findall(r'([\u4e00-\u9fa5]+站)', text)
if len(stations) >= 2:
real_stations = [s for s in stations if len(s) >= 3 and '站' in s and '12306' not in s]
if len(real_stations) >= 2:
result["category"] = f"铁路客运-{real_stations[0]}→{real_stations[-1]}"
elif len(real_stations) == 1:
result["category"] = f"铁路客运-{real_stations[0]}"
elif len(stations) == 1:
result["category"] = f"铁路客运-{stations[0]}"
logging.info(f"铁路客票-提取结果: 销售方={result['seller_name']}, 金额={result['amount']}, 类别={result['category']}, 税号={result['tax_id']}")
return result
def normalize_date(date_str):
if not date_str:
return None
date_str = str(date_str).strip()
m = re.match(r'(\d{4})-(\d{1,2})-(\d{1,2})', date_str)
if m:
try:
return datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
except ValueError:
return None
m = re.search(r'(\d{4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*日', date_str)
if m:
try:
return datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
except ValueError:
return None
m = re.match(r'(\d{4})/(\d{1,2})/(\d{1,2})', date_str)
if m:
try:
return datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
except ValueError:
return None
return None
def is_airline_ticket(text):
return '航空运输电子客票行程单' in text or '航空运输' in text
def extract_airline_ticket_data(text):
result = {
"is_airline": False,
"seller_name": "",
"tax_id": "",
"buyer_name": "",
"buyer_tax_id": "",
"amount": 0.0,
"tax_amount": 0.0,
"total_amount": 0.0,
"invoice_number": "",
"date": None,
"category": "航空客运",
"tax_rate": 0.0
}
if not is_airline_ticket(text):
return result
result["is_airline"] = True
logging.info("检测到航空客票格式")
result["tax_id"] = "航空客票无税号"
invoice_match = re.search(r'发票号码[：:]\s*(\d+)', text)
if invoice_match:
result["invoice_number"] = invoice_match.group(1)
seller_match = re.search(r'填开单位[：:][ \t]*([^\n]+)', text)
if seller_match:
seller_name = seller_match.group(1).strip()
seller_name = re.sub(r'\s*填开日期.*', '', seller_name)
result["seller_name"] = seller_name
date_match = re.search(r'填开日期[：:]\s*(\d{4}年\d{1,2}月\d{1,2}日)', text)
if date_match:
result["date"] = normalize_date(date_match.group(1))
lines = text.split('\n')
cny_value_lines = []
for i, line in enumerate(lines):
if 'CNY' in line and re.search(r'CNY\s*\d+', line):
cny_value_lines.append((i, line))
all_cny_values = []
for line_idx, line in cny_value_lines:
cny_matches = re.findall(r'CNY\s*(\d+\.?\d*)', line)
for v in cny_matches:
try:
all_cny_values.append(float(v))
except:
pass
fare = 0.0
fuel = 0.0
fund = 0.0
tax = 0.0
total = 0.0
if len(all_cny_values) >= 5:
fare = all_cny_values[0]
fuel = all_cny_values[1]
tax = all_cny_values[2]
fund = all_cny_values[3]
total = all_cny_values[-1]
elif len(all_cny_values) >= 4:
fare = all_cny_values[0]
fuel = all_cny_values[1]
tax = all_cny_values[2]
total = all_cny_values[-1]
elif len(all_cny_values) >= 3:
fare = all_cny_values[0]
tax = all_cny_values[1] if all_cny_values[1] < all_cny_values[0] else 0
total = all_cny_values[-1]
elif len(all_cny_values) >= 2:
fare = all_cny_values[0]
total = all_cny_values[-1]
elif len(all_cny_values) >= 1:
total = all_cny_values[0]
airline_tax_rate = 0.0
for line in lines:
if 'CNY' in line:
rate_match = re.search(r'(\d+)\.?\d*\s*%', line)
if rate_match:
try:
airline_tax_rate = float(rate_match.group(1))
except:
pass
break
result["tax_amount"] = tax
result["total_amount"] = total
result["amount"] = total - tax if total > tax else fare
buyer_match = re.search(r'购买方名称[：:]\s*([^\n]+)', text)
if buyer_match:
buyer_text = buyer_match.group(1).strip()
parts = re.split(r'\s+统一社会信用代码[//]纳税人识别号[：:]', buyer_text)
if len(parts) >= 2:
result["buyer_name"] = parts[0].strip()
result["buyer_tax_id"] = parts[1].strip()
else:
result["buyer_name"] = buyer_text
result["tax_rate"] = airline_tax_rate
logging.info(f"航空客票最终结果: 不含税金额={result['amount']}, 税额={result['tax_amount']}, 价税合计={result['total_amount']}, 税率={airline_tax_rate}%")
return result
def is_travel_itinerary(text):
return '行程单' in text and '航空运输电子客票行程单' not in text
def extract_travel_itinerary_data(text):
result = {
"is_itinerary": False,
"seller_name": "",
"tax_id": "",
"amount": 0.0,
"tax_amount": 0.0,
"total_amount": 0.0,
"invoice_number": "",
"date": None,
"category": "行程单"
}
if not is_travel_itinerary(text):
return result
result["is_itinerary"] = True
logging.info("检测到行程单格式")
result["tax_id"] = "行程单无税号"
invoice_match = re.search(r'(?:发票号码|行程单号|票据号码)[：:]\s*(\d+)', text)
if invoice_match:
result["invoice_number"] = invoice_match.group(1)
seller_match = re.search(r'(?:填开单位|承运人|出票单位)[：:][ \t]*([^\n]+)', text)
if seller_match:
seller_name = seller_match.group(1).strip()
seller_name = re.sub(r'\s*填开日期.*', '', seller_name)
result["seller_name"] = seller_name
date_match = re.search(r'(?:填开日期|开票日期|日期)[：:]\s*(\d{4}年\d{1,2}月\d{1,2}日)', text)
if not date_match:
date_match = re.search(r'(?:填开日期|开票日期|日期)[：:]\s*(\d{4}-\d{1,2}-\d{1,2})', text)
if date_match:
result["date"] = normalize_date(date_match.group(1))
lines = text.split('\n')
cny_values = []
for line in lines:
if 'CNY' in line:
cny_matches = re.findall(r'CNY\s*(\d+\.?\d*)', line)
for v in cny_matches:
try:
cny_values.append(float(v))
except:
pass
if cny_values:
result["total_amount"] = cny_values[-1]
if len(cny_values) >= 3:
result["tax_amount"] = cny_values[2]
elif len(cny_values) >= 2:
result["tax_amount"] = cny_values[-2] if cny_values[-2] < cny_values[-1] else 0
result["amount"] = result["total_amount"] - result["tax_amount"]
else:
amounts = extract_amounts(text)
result["amount"] = amounts["amount"]
result["tax_amount"] = amounts["tax_amount"]
result["total_amount"] = amounts["total_amount"]
logging.info(f"行程单-提取结果: 销售方={result['seller_name']}, 金额={result['amount']}, 税额={result['tax_amount']}, 合计={result['total_amount']}")
return result
# ================== 核心提取函数 ==================
def extract_seller_name(text):
"""提取销售方名称 - 包含块匹配、两列布局提取、跨行合并"""
# ----- 新增：块匹配（优先处理“销售方信息”和“购买方信息”格式） -----
clean_text = re.sub(r'<[^>]+>', '', text) # 去除HTML标签
def clean_name(name):
if not name:
return ""
name = name.strip()
name = re.sub(r'\s+', '', name)
name = re.sub(r'^名称[:：]?', '', name)
name = re.sub(r'^销售方名称[:：]?', '', name)
name = re.sub(r'^购买方名称[:：]?', '', name)
name = re.sub(r'^售名称[:：]?', '', name)
name = re.sub(r'^销名称[:：]?', '', name)
name = re.sub(r'^购名称[:：]?', '', name)
name = re.sub(r'^[:：]', '', name)
name = re.sub(r'[<>:"/\\|?*]', '', name)
name = re.sub(r'[\d\s\-_;；,，]+$', '', name)
return name
seller_name = ""
buyer_name = ""
# 1. 从“销售方信息”块提取
seller_block = re.search(r'销售方信息.*?名称[：:]\s*([^\n]+)', clean_text, re.IGNORECASE | re.DOTALL)
if seller_block:
candidate = seller_block.group(1).strip()
for kw in ['统一社会信用代码', '纳税人识别号']:
if kw in candidate:
candidate = candidate.split(kw)[0].strip()
break
candidate = clean_name(candidate)
if candidate and len(candidate) >= 2:
seller_name = candidate
logging.info(f'块匹配-销售方: {seller_name}')
# 2. 从“购买方信息”块提取
buyer_block = re.search(r'购买方信息.*?名称[：:]\s*([^\n]+)', clean_text, re.IGNORECASE | re.DOTALL)
if buyer_block:
candidate = buyer_block.group(1).strip()
for kw in ['统一社会信用代码', '纳税人识别号']:
if kw in candidate:
candidate = candidate.split(kw)[0].strip()
break
candidate = clean_name(candidate)
if candidate and len(candidate) >= 2:
buyer_name = candidate
logging.info(f'块匹配-购买方: {buyer_name}')
# 如果块匹配成功，直接返回
if seller_name and buyer_name:
logging.info(f'名称提取结果: 销售方={seller_name}, 购买方={buyer_name}')
return {"seller": seller_name, "buyer": buyer_name}
# ----- 以下是原有逻辑（两列布局、跨行合并等） -----
lines = text.split('\n')
def merge_cross_line_company(company, start_line_idx):
if start_line_idx + 1 >= len(lines):
return company
merged = company
exclude_keywords = ['银行', '账号', '地址', '电话', '传真', '开户', '税号',
'纳税人', '开户行', '收款', '复核', '开票', '销售方',
'购买方', '价税', '金额', '备注', '规格', '单位', '数量',
'税率', '税额', '大写', '小写', '合计', '货物', '服务',
'名称', '统一社会信用代码', '代码', '密文', '校验码',
'买售', '买方', '购方', '购买', '地址电话']
company_endings = ['公司', '站', '店', '厂', '院', '中心', '部', '所', '行',
'有限', '股份', '集团', '油站', '加油站', '服务区']
for offset in range(1, min(9, len(lines) - start_line_idx)):
next_line = lines[start_line_idx + offset].strip()
if not next_line:
continue
next_line_no_space = next_line.replace(' ', '')
if any(kw in next_line for kw in exclude_keywords):
break
if len(next_line_no_space) <= 15:
if re.match(r'^[\u4e00-\u9fa5\d（）\(\)]+$', next_line_no_space):
test_merged = merged + next_line_no_space
should_continue = any(next_line_no_space.endswith(end) for end in company_endings)
if any(merged.endswith(end) for end in company_endings) and len(next_line_no_space) > 2:
if len(next_line_no_space) <= 2:
merged = test_merged
continue
break
merged = test_merged
if should_continue:
continue
else:
break
break
return merged
buyer_name_orig = ""
buyer_patterns = [
r'(?:购买方|购方|买方)\s*名称[：:][ \t]*([^\n销售]+)',
r'名称[：:][ \t]*([^\n]+?)(?:\s*[销售]|$)',
r'买\s*名称[：:][ \t]*([^\n销售]+)',
]
for pattern in buyer_patterns:
match = re.search(pattern, text)
if match:
buyer_name_orig = match.group(1).strip()
buyer_name_orig = re.sub(r'\s*(银行|账号|：|:).*', '', buyer_name_orig)
break
# 方法1: 精确匹配"销/售 名称："或"销/售名称："
pattern1 = re.search(r'[销售]\s*名称[：:][ \t]*([^\n]*)', text)
if pattern1:
candidate = pattern1.group(1).strip()
candidate = re.split(r'[;；\n]', candidate)[0].strip()
if not candidate or len(candidate) < 2:
match_start = pattern1.start()
line_idx = 0
char_count = 0
for i, line in enumerate(lines):
char_count += len(line) + 1
if char_count > match_start:
line_idx = i
break
for offset in range(1, min(31, len(lines) - line_idx)):
check_line = lines[line_idx + offset].strip()
companies_on_line = re.findall(r'([一-龥（）\(\)]{4,}(?:公司|有限|加油站|服务区|酒店|商店))', check_line)
if len(companies_on_line) >= 2:
candidate = companies_on_line[-1]
if not buyer_name_orig:
buyer_name_orig = companies_on_line[0]
logging.info(f'从两列布局行提取销售方: {candidate}, 购买方: {buyer_name_orig}')
if offset + 1 < min(31, len(lines) - line_idx):
tax_line = lines[line_idx + offset + 1].strip()
tax_codes = re.findall(r'([A-Z0-9]{15,20})', tax_line)
if len(tax_codes) >= 2:
logging.info(f'从两列税号行提取: 购买方={tax_codes[0]}, 销售方={tax_codes[-1]}')
break
if '银行' not in candidate and '账号' not in candidate and len(candidate) >= 2:
match_start = pattern1.start()
line_idx = 0
char_count = 0
for i, line in enumerate(lines):
char_count += len(line) + 1
if char_count > match_start:
line_idx = i
break
if line_idx + 1 < len(lines):
next_line = lines[line_idx + 1].strip()
if '买' in next_line and '售' in next_line:
right_col_match = re.search(r'售\s*([^\s]+(?:\s+[^\s]+)*?)$', next_line)
if right_col_match:
right_content = right_col_match.group(1).strip()
if right_content and len(right_content) <= 15:
right_content_no_space = right_content.replace(' ', '')
if re.match(r'^[\u4e00-\u9fa5\d（）\(\)]+$', right_content_no_space):
candidate = candidate + right_content_no_space
seller_name = candidate
# 方法2: 匹配"销售方名称："或"销方名称："或"售方名称："
if not seller_name:
pattern2 = re.search(r'(?:销售方|销方|售方)\s*名称[：:][ \t]*([^\n]*)', text)
if pattern2:
candidate = pattern2.group(1).strip()
candidate = re.split(r'[;；\n]', candidate)[0].strip()
if '银行' not in candidate and '账号' not in candidate:
match_start = pattern2.start()
line_idx = 0
char_count = 0
for i, line in enumerate(lines):
char_count += len(line) + 1
if char_count > match_start:
line_idx = i
break
seller_name = merge_cross_line_company(candidate, line_idx)
# 方法3: 找"名称：购买方 xxx 销 名称：销售方 xxx"这种格式
if not seller_name:
name_matches = list(re.finditer(r'名称[：:][ \t]*([^\n]+)', text))
for i, match in enumerate(name_matches):
full_match_text = match.group(0)
matched_value = match.group(1).strip()
if ('销' in full_match_text or '售' in full_match_text) and '购买' not in full_match_text:
seller_marker = '销' if '销' in full_match_text else '售'
pos_in_match = full_match_text.find(seller_marker)
if pos_in_match < full_match_text.find(matched_value[:10] if len(matched_value) > 10 else matched_value):
candidate = matched_value
else:
split_parts = re.split(r'\s*[销售]\s*名称[：:]\s*', matched_value)
if len(split_parts) > 1:
candidate = split_parts[-1].strip()
else:
after_seller_marker = re.search(r'[销售]\s*名称[：:]\s*(.+)', full_match_text)
if after_seller_marker:
candidate = after_seller_marker.group(1).strip()
else:
candidate = matched_value
candidate = re.split(r'[;；\n]', candidate)[0].strip()
candidate = re.sub(r'\s*(银行|账号|：|:).*', '', candidate)
if re.search(r'(公司|有限|加油站|石油|服务区|能源|贸易|石化|油站|股份|超市|商店)', candidate):
if '银行' not in candidate and '账号' not in candidate:
match_start = match.start()
line_idx = 0
char_count = 0
for j, line in enumerate(lines):
char_count += len(line) + 1
if char_count > match_start:
line_idx = j
break
seller_name = merge_cross_line_company(candidate, line_idx)
break
# 方法4: 根据文本位置判断（销售方在购买方后面）
if not seller_name:
buyer_pos = max(
text.find('购买方'),
text.find('购方'),
text.find('买方')
)
seller_pos = max(
text.find('销售方'),
text.find('销方'),
text.find('售方'),
text.find('\n销'),
text.find('\n售')
)
if seller_pos > buyer_pos and buyer_pos >= 0:
text_after_seller = text[seller_pos:]
company_match = re.search(r'([\u4e00-\u9fa5]{2,}(?:加油站|石油|服务区|能源|贸易|石化|油站|股份|有限|公司|超市|商店)[^\n;；]*)', text_after_seller)
if company_match:
candidate = company_match.group(1).strip()
candidate = re.sub(r'\s*(银行|账号).*', '', candidate)
if '银行' not in candidate and '账号' not in candidate:
match_start = seller_pos + company_match.start()
line_idx = 0
char_count = 0
for j, line in enumerate(lines):
char_count += len(line) + 1
if char_count > match_start:
line_idx = j
break
seller_name = merge_cross_line_company(candidate, line_idx)
# 方法5: 从发票下半部分找公司名
if not seller_name:
half_text = text[len(text)//2:]
companies = re.findall(r'([\u4e00-\u9fa5]{2,}(?:加油站|石油|服务区|能源|贸易|石化|油站|股份|有限|公司|超市|商店)[^\n;；]*)', half_text)
valid_companies = []
for company in companies:
company = company.strip()
company = re.sub(r'\s*(银行|账号).*', '', company)
if buyer_name_orig and buyer_name_orig in company:
continue
if '银行' in company or '账号' in company:
continue
if '购买' in company or '购方' in company or '买方' in company:
continue
if company:
valid_companies.append(company)
if valid_companies:
best_company = max(valid_companies, key=len)
company_pos = text.find(best_company)
if company_pos >= 0:
line_idx = 0
char_count = 0
for j, line in enumerate(lines):
char_count += len(line) + 1
if char_count > company_pos:
line_idx = j
break
seller_name = merge_cross_line_company(best_company, line_idx)
else:
seller_name = best_company
# 清理名称
if seller_name:
seller_name = re.sub(r'\s+', '', seller_name)
seller_name = re.sub(r'买售.*$', '', seller_name)
seller_name = re.sub(r'方方.*$', '', seller_name)
seller_name = re.sub(r'^名称[：:]', '', seller_name)
seller_name = re.sub(r'^销售方', '', seller_name)
seller_name = re.sub(r'^销方', '', seller_name)
seller_name = re.sub(r'^销\s*', '', seller_name)
seller_name = re.sub(r'^\d+\s*', '', seller_name)
seller_name = re.sub(r'[\d\s\-_;；,，]+$', '', seller_name)
seller_name = re.sub(r'[<>:"/\\|?*]', '', seller_name)
if len(seller_name) > 80:
seller_name = seller_name[:80]
# 如果块匹配未找到购买方，但原有逻辑找到了，使用原有逻辑的购买方
if not buyer_name and buyer_name_orig:
buyer_name = buyer_name_orig
return {"seller": seller_name if seller_name else "未识别", "buyer": buyer_name if buyer_name else ""}
def extract_seller_tax_id(text):
tax_id = ""
buyer_tax_id = ""
lines = text.split('\n')
seller_tax_block = re.search(r'销售方信息.*?统一社会信用代码[：/]*\s*([A-Z0-9]{15,20})', text, re.IGNORECASE | re.DOTALL)
if seller_tax_block:
tax_id = seller_tax_block.group(1).strip()
logging.info(f'从销售方信息块提取税号: {tax_id}')
buyer_tax_block = re.search(r'购买方信息.*?统一社会信用代码[：/]*\s*([A-Z0-9]{15,20})', text, re.IGNORECASE | re.DOTALL)
if buyer_tax_block:
buyer_tax_id = buyer_tax_block.group(1).strip()
logging.info(f'从购买方信息块提取税号: {buyer_tax_id}')
if not tax_id and not buyer_tax_id:
for line in lines:
if '统一社会信用代码' in line or '纳税人识别号' in line:
tax_ids = re.findall(r'([A-Z0-9]{15,20})', line)
if len(tax_ids) >= 2:
tax_id = tax_ids[-1]
buyer_tax_id = tax_ids[0]
logging.info(f'两列布局税号，购买方: {buyer_tax_id}, 销售方: {tax_id}')
break
elif len(tax_ids) == 1:
line_idx = lines.index(line)
for i in range(max(0, line_idx - 5), line_idx):
if ('销' in lines[i] or '售' in lines[i]) and '购' not in lines[i]:
tax_id = tax_ids[0]
break
if tax_id:
break
if not tax_id:
seller_match = re.search(r'[销售]\s*名称[：:][ \t]*([^\n]*)', text)
if seller_match:
seller_pos = seller_match.end()
text_after_seller = text[seller_pos:seller_pos + 500]
tax_match = re.search(r'(?:统一社会信用代码|纳税人识别号)[：:/]*\s*([A-Z0-9]{15,20})', text_after_seller)
if tax_match:
tax_id = tax_match.group(1)
logging.info(f'从销售方名称后查找税号: {tax_id}')
if not tax_id:
all_tax_ids = re.findall(r'(?:统一社会信用代码|纳税人识别号)[：:/]*\s*([A-Z0-9]{15,20})', text)
if len(all_tax_ids) >= 2:
tax_id = all_tax_ids[-1]
if not buyer_tax_id:
buyer_tax_id = all_tax_ids[0]
elif len(all_tax_ids) == 1:
tax_id = all_tax_ids[0]
if not tax_id:
credit_codes = re.findall(r'(?<!\d)([A-Z0-9]{18})(?!\d)', text)
if credit_codes:
tax_id = credit_codes[-1]
if len(credit_codes) >= 2 and not buyer_tax_id:
buyer_tax_id = credit_codes[0]
return {"seller": tax_id if tax_id else "未识别", "buyer": buyer_tax_id}
def extract_amounts(text):
lines = text.split('\n')
total_amount = 0.0
tax_amount = 0.0
amount = 0.0
total_patterns = [
r'价税合计[（(]小写[）)]?\s*[￥]?\s*(\d+\.?\d*)',
r'价税合计\s*[￥]?\s*(\d+\.?\d*)',
r'价 税 合 计\s*[￥]?\s*(\d+\.?\d*)',
r'小写\s*[￥]?\s*(\d+\.?\d*)',
]
for pattern in total_patterns:
match = re.search(pattern, text)
if match:
val = float(match.group(1))
if 0 < val < 1_000_000:
total_amount = round(val, 2)
logging.info(f'提取价税合计: {total_amount}')
break
tax_patterns = [
r'税额\s*[￥]?\s*(\d+\.?\d*)',
r'税 额\s*[￥]?\s*(\d+\.?\d*)',
]
for pattern in tax_patterns:
match = re.search(pattern, text)
if match:
val = float(match.group(1))
if 0 < val < 1_000_000:
tax_amount = round(val, 2)
logging.info(f'提取税额: {tax_amount}')
break
for m in re.finditer(r'金额\s*[￥]?\s*(\d+\.?\d*)', text):
start = m.start()
if start >= 6 and '价税合计' in text[start-6:start]:
continue
val = float(m.group(1))
if 0 < val < 1_000_000:
amount = round(val, 2)
logging.info(f'提取金额: {amount}')
break
if amount == 0:
m = re.search(r'金 额\s*[￥]?\s*(\d+\.?\d*)', text)
if m:
val = float(m.group(1))
if 0 < val < 1_000_000:
amount = round(val, 2)
logging.info(f'提取金额: {amount}')
if total_amount == 0 or tax_amount == 0 or amount == 0:
for i, line in enumerate(lines):
if '合计' in line or '合 计' in line:
numbers = []
for offset in range(-2, 3):
idx = i + offset
if 0 <= idx < len(lines):
nums = re.findall(r'\b(\d+\.\d{2})\b', lines[idx])
numbers.extend([float(n) for n in nums if 0 < float(n) < 1_000_000])
numbers = sorted(set(numbers), reverse=True)
logging.info(f'从合计行提取金额: {numbers[0] if numbers else 0}')
if len(numbers) >= 3 and total_amount == 0:
if numbers[0] > numbers[1] > numbers[2]:
total_amount = numbers[0]
amount = numbers[1]
tax_amount = numbers[2]
elif len(numbers) == 2 and total_amount == 0:
if numbers[0] > numbers[1]:
total_amount = numbers[0]
tax_amount = numbers[1]
amount = total_amount - tax_amount
else:
total_amount = numbers[1]
tax_amount = numbers[0]
amount = total_amount - tax_amount
elif len(numbers) == 1 and total_amount == 0:
total_amount = numbers[0]
break
if amount == 0 and tax_amount == 0 and total_amount > 0:
for line in lines:
if '*' in line and re.search(r'\d+\.\d{2}', line):
nums = re.findall(r'\b(\d+\.\d{2})\b', line)
if len(nums) >= 2:
vals = [float(n) for n in nums if 0 < float(n) < 1_000_000]
if len(vals) >= 2:
amount = vals[-2]
tax_amount = vals[-1]
if abs(amount + tax_amount - total_amount) > 0.1 and total_amount > 0:
if abs(tax_amount + amount - total_amount) <= 0.1:
amount, tax_amount = tax_amount, amount
else:
if total_amount > tax_amount:
amount = total_amount - tax_amount
elif total_amount > amount:
tax_amount = total_amount - amount
break
if total_amount > 0 and amount > 0 and tax_amount > 0:
if abs(amount + tax_amount - total_amount) > 0.01:
if abs(tax_amount + amount - total_amount) <= 0.01:
amount, tax_amount = tax_amount, amount
logging.info(f'交换金额和税额后成立')
else:
new_amount = total_amount - tax_amount
if new_amount > 0 and abs(new_amount + tax_amount - total_amount) <= 0.01:
amount = new_amount
logging.info(f'用总价-税额修正金额')
else:
new_tax = total_amount - amount
if new_tax > 0 and abs(amount + new_tax - total_amount) <= 0.01:
tax_amount = new_tax
logging.info(f'用总价-金额修正税额')
elif total_amount > 0 and amount == 0 and tax_amount > 0:
amount = total_amount - tax_amount
logging.info(f'由总价和税额计算金额: {amount}')
elif total_amount > 0 and amount > 0 and tax_amount == 0:
tax_amount = total_amount - amount
logging.info(f'由总价和金额计算税额: {tax_amount}')
if amount == 0 and total_amount > 0:
amount = total_amount
logging.info(f'将总价作为金额: {amount}')
amount = round(amount, 2)
tax_amount = round(tax_amount, 2)
total_amount = round(total_amount, 2)
if amount > 0 and tax_amount > 0:
tax_rate = round(tax_amount / amount * 100, 2)
else:
tax_rate = 0.0
logging.info(f'金额提取: 金额={amount}, 税额={tax_amount}, 价税合计={total_amount}, 税率={tax_rate}%')
return {
"amount": amount,
"tax_amount": tax_amount,
"total_amount": total_amount,
"tax_rate": tax_rate
}
def read_pdf_content(file_path):
logging.info(f'开始处理文件: {os.path.basename(file_path)}')
try:
with pdfplumber.open(file_path) as pdf:
full_text = ""
for page in pdf.pages:
page_text = page.extract_text()
if page_text:
full_text += page_text + "\n"
logging.info(f'提取文本长度: {len(full_text)}')
if len(full_text.strip()) == 0:
logging.warning(f'扫描版PDF（无文字信息），无法提取数据: {os.path.basename(file_path)}')
patent_data = extract_patent_fee_data(full_text)
if patent_data["is_patent"]:
return {
"amount": patent_data["amount"],
"tax_amount": patent_data["tax_amount"],
"total_amount": patent_data["total_amount"],
"tax_rate": 0.0,
"invoice_number": patent_data["invoice_number"],
"name": patent_data["seller_name"],
"tax_id": patent_data["tax_id"],
"buyer_name": "",
"buyer_tax_id": "",
"date": patent_data["date"],
"category": patent_data["category"],
"project_name": ""
}
railway_data = extract_railway_ticket_data(full_text)
if railway_data["is_railway"]:
return {
"amount": railway_data["amount"],
"tax_amount": railway_data["tax_amount"],
"total_amount": railway_data["total_amount"],
"tax_rate": 0.0,
"invoice_number": railway_data["invoice_number"],
"name": railway_data["seller_name"],
"tax_id": railway_data["tax_id"],
"buyer_name": railway_data.get("buyer_name", ""),
"buyer_tax_id": railway_data.get("buyer_tax_id", ""),
"date": railway_data["date"],
"category": railway_data["category"],
"project_name": ""
}
airline_data = extract_airline_ticket_data(full_text)
if airline_data["is_airline"]:
return {
"amount": airline_data["amount"],
"tax_amount": airline_data["tax_amount"],
"total_amount": airline_data["total_amount"],
"tax_rate": airline_data.get("tax_rate", 0.0),
"invoice_number": airline_data["invoice_number"],
"name": airline_data["seller_name"],
"tax_id": airline_data["tax_id"],
"buyer_name": airline_data.get("buyer_name", ""),
"buyer_tax_id": airline_data.get("buyer_tax_id", ""),
"date": airline_data["date"],
"category": airline_data["category"],
"project_name": ""
}
itinerary_data = extract_travel_itinerary_data(full_text)
if itinerary_data["is_itinerary"]:
itin_tax_rate = round(itinerary_data["tax_amount"] / itinerary_data["amount"] * 100, 2) if itinerary_data["amount"] > 0 and itinerary_data["tax_amount"] > 0 else 0.0
return {
"amount": itinerary_data["amount"],
"tax_amount": itinerary_data["tax_amount"],
"total_amount": itinerary_data["total_amount"],
"tax_rate": itin_tax_rate,
"invoice_number": itinerary_data["invoice_number"],
"name": itinerary_data["seller_name"],
"tax_id": itinerary_data["tax_id"],
"buyer_name": "",
"buyer_tax_id": "",
"date": itinerary_data["date"],
"category": itinerary_data["category"],
"project_name": ""
}
invoice_number = re.findall(r'(?:发票号码|发票代码|发票号)[\s]*[：:]*\s*([0-9]{8,20})', full_text)
if not invoice_number:
invoice_number = re.findall(r'([0-9]{20})', full_text)
seller_info = extract_seller_name(full_text)
seller_name = seller_info["seller"]
buyer_name = seller_info["buyer"]
tax_info = extract_seller_tax_id(full_text)
seller_tax_id = tax_info["seller"]
buyer_tax_id = tax_info["buyer"]
date = re.findall(r'(\d{4}\s*年\s*\d{1,2}\s*月\s*\d{1,2}\s*日)', full_text)
if not date:
date = re.findall(r'(\d{4}-\d{1,2}-\d{1,2})', full_text)
if not date:
date = re.findall(r'(\d{4}年\d{1,2}月\d{1,2}日)', full_text)
if not date:
date = re.findall(r'(\d{4}/\d{1,2}/\d{1,2})', full_text)
category = ""
project_name = ""
goods_pair = re.search(r'\*([^*]+)\*', full_text)
if goods_pair:
category = goods_pair.group(1).strip()
rest = full_text[goods_pair.end():].lstrip()
end_match = re.search(r'[\s\n\r*]', rest)
if end_match:
project_name = rest[:end_match.start()].strip()
else:
project_name = rest.strip()
logging.info(f'提取项目(成对匹配): 分类={category}, 项目={project_name}')
if not category:
goods_matches = re.findall(r'\*([^*]+)\*', full_text)
if goods_matches:
category = goods_matches[0].strip()
project_name = category
logging.info(f'提取项目(单匹配): 分类={category}')
if not category:
service_match = re.search(r'货物或应税劳务、服务名称[^\n]*\n([^\n]+)', full_text)
if service_match:
category = service_match.group(1).strip()
project_name = category
amounts = extract_amounts(full_text)
logging.info(f'成功提取: 金额={amounts["amount"]}, 税额={amounts["tax_amount"]}, 价税合计={amounts["total_amount"]}, 税率={amounts["tax_rate"]}%, 发票号码={invoice_number[0] if invoice_number else "未找到"}, 销售方={seller_name}, 销售方税号={seller_tax_id}, 购买方={buyer_name}')
except Exception as e:
logging.error(f"读取PDF文件失败 {file_path}: {e}")
return {
"amount": 0.0,
"tax_amount": 0.0,
"total_amount": 0.0,
"tax_rate": 0.0,
"invoice_number": "",
"name": "读取失败",
"tax_id": "",
"buyer_name": "",
"buyer_tax_id": "",
"date": None,
"category": "",
"project_name": ""
}
final_amount = amounts["amount"]
final_total = amounts["total_amount"]
if final_amount == 0 and final_total > 0:
final_amount = final_total
return {
"amount": final_amount,
"tax_amount": amounts["tax_amount"],
"total_amount": amounts["total_amount"],
"tax_rate": amounts["tax_rate"],
"invoice_number": invoice_number[0] if invoice_number else "",
"name": seller_name,
"tax_id": seller_tax_id,
"buyer_name": buyer_name,
"buyer_tax_id": buyer_tax_id,
"date": normalize_date(date[0]) if date else None,
"category": category,
"project_name": project_name
}
def get_pdf_files(pdf_dir):
logging.info('开始扫描PDF文件')
pdf_files = []
for root, dirs, files in os.walk(pdf_dir):
for file in files:
if file.lower().endswith(".pdf"):
filepath = os.path.normpath(os.path.join(root, file))
pdf_files.append(filepath)
logging.info(f'找到 {len(pdf_files)} 个PDF文件')
return pdf_files
def rename_pdf_file(file_path, new_value):
logging.info(f'重命名文件: {file_path}')
dir_path = os.path.dirname(file_path)
invalid_chars = '<>:"/\\|?*'
for char in invalid_chars:
new_value = new_value.replace(char, '_')
if len(new_value) > 200:
new_value = new_value[:200]
new_file_name = f"{new_value}.pdf"
new_file_path = os.path.join(dir_path, new_file_name)
counter = 1
while os.path.exists(new_file_path):
new_file_name = f"{new_value}_{counter}.pdf"
new_file_path = os.path.join(dir_path, new_file_name)
counter += 1
os.rename(file_path, new_file_path)
logging.info(f'重命名完成: {new_file_path}')
return new_file_path
def open_pdf(path):
if os.name == 'nt':
os.startfile(path)
else:
opener = 'open' if os.name == 'posix' else 'xdg-open'
subprocess.call([opener, path])
def center_window(root, width=600, height=250):
screen_width = root.winfo_screenwidth()
screen_height = root.winfo_screenheight()
x = (screen_width - width) // 2
y = (screen_height - height) // 2
root.geometry(f"{width}x{height}+{x}+{y}")
root.minsize(width, height)
# ================== UI ==================
def display_results(values, total_amount_sum, total_tax_sum, total_sum, input_root, pdf_source_folder):
input_root.destroy()
raw_values = list(values)
root = tk.Tk()
root.title("发票金额统计结果")
center_window(root, width=1700, height=700)
root.minsize(1200, 600)
main_frame = ttk.Frame(root, padding="10")
main_frame.grid(column=0, row=0, sticky=(tk.W, tk.E, tk.N, tk.S))
main_frame.columnconfigure(0, weight=1)
main_frame.rowconfigure(0, weight=1)
root.columnconfigure(0, weight=1)
root.rowconfigure(0, weight=1)
def rename_selected_files():
selected_items = tree.selection()
if not selected_items:
messagebox.showerror("错误", "请先选择一个或多个PDF文件进行重命名")
return
dialog = Toplevel(root)
dialog.title("选择需要的字段")
dialog.geometry("350x500")
dialog.transient(root)
dialog.grab_set()
include_amount = StringVar(value='no')
include_tax = StringVar(value='no')
include_total = StringVar(value='yes')
include_project_name = StringVar(value='no')
include_invoice_number = StringVar(value='yes')
include_name = StringVar(value='yes')
include_tax_id = StringVar(value='no')
include_buyer_name = StringVar(value='no')
include_buyer_tax_id = StringVar(value='no')
include_date = StringVar(value='no')
frame = ttk.Frame(dialog, padding="20")
frame.pack(fill='both', expand=True)
ttk.Label(frame, text="请选择要包含在文件名中的字段：", font=('Arial', 10, 'bold')).pack(pady=(0, 10))
Checkbutton(frame, text='金额（不含税）', variable=include_amount, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='税额', variable=include_tax, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='价税合计', variable=include_total, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='项目名称', variable=include_project_name, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='发票号码', variable=include_invoice_number, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='销售方名称', variable=include_name, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='销售方纳税人识别号', variable=include_tax_id, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='购买方名称', variable=include_buyer_name, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='购买方纳税人识别号', variable=include_buyer_tax_id, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
Checkbutton(frame, text='开票日期', variable=include_date, onvalue='yes', offvalue='no', anchor='w').pack(fill='x', pady=3)
def on_ok():
renamed_count = 0
for item in selected_items:
item_values = tree.item(item, 'values')
new_name_parts = []
if include_amount.get() == 'yes' and item_values[1]:
new_name_parts.append(f"金额{item_values[1]}")
if include_tax.get() == 'yes' and item_values[3]:
new_name_parts.append(f"税额{item_values[3]}")
if include_total.get() == 'yes' and item_values[4]:
new_name_parts.append(f"{item_values[4]}")
if include_project_name.get() == 'yes' and item_values[10]:
new_name_parts.append(str(item_values[10]))
if include_invoice_number.get() == 'yes' and item_values[5]:
new_name_parts.append(str(item_values[5]))
if include_name.get() == 'yes' and item_values[6] and item_values[6] != "未识别":
new_name_parts.append(str(item_values[6]))
if include_tax_id.get() == 'yes' and item_values[7] and item_values[7] != "未识别":
new_name_parts.append(str(item_values[7]))
if include_buyer_name.get() == 'yes' and item_values[8] and item_values[8]:
new_name_parts.append(str(item_values[8]))
if include_buyer_tax_id.get() == 'yes' and item_values[9] and item_values[9]:
new_name_parts.append(str(item_values[9]))
if include_date.get() == 'yes' and item_values[11]:
new_name_parts.append(str(item_values[11]))
if not new_name_parts:
continue
new_file_name = "_".join(new_name_parts)
current_file_path = item_values[12]
new_file_path = rename_pdf_file(current_file_path, new_file_name)
tree.set(item, column="文件路径", value=new_file_path)
renamed_count += 1
messagebox.showinfo("完成", f"成功重命名 {renamed_count} 个文件。")
dialog.destroy()
Button(frame, text='确定', command=on_ok, width=15).pack(pady=20)
def export_to_xls():
save_path = filedialog.asksaveasfilename(
title="保存Excel文件",
defaultextension=".xls",
filetypes=[("Excel files", "*.xls"), ("All files", "*.*")],
initialfile="发票数据.xls",
initialdir=pdf_source_folder
)
if not save_path:
return
try:
workbook = xlwt.Workbook(encoding='utf-8')
sheet = workbook.add_sheet('发票数据')
col_widths = [8, 15, 10, 15, 18, 25, 30, 25, 30, 25, 20, 15, 50]
for i, width in enumerate(col_widths):
sheet.col(i).width = 256 * width
style = xlwt.easyxf('align: vert centre, horiz centre')
style_text = xlwt.easyxf('align: vert centre, horiz left')
style_special = xlwt.easyxf('align: vert centre, horiz centre; pattern: pattern solid, fore_colour light_yellow')
style_special_text = xlwt.easyxf('align: vert centre, horiz left; pattern: pattern solid, fore_colour light_yellow')
style_date = xlwt.easyxf('align: vert centre, horiz centre', num_format_str='yyyy-mm-dd')
style_date_special = xlwt.easyxf('align: vert centre, horiz centre; pattern: pattern solid, fore_colour light_yellow', num_format_str='yyyy-mm-dd')
headers = ["序号", "金额(不含税)", "税率", "税额", "价税合计", "发票号码", "销售方名称", "销售方税号", "购买方名称", "购买方税号", "项目名称", "开票日期", "文件路径"]
for i, header in enumerate(headers):
sheet.write(0, i, header, style)
for i, item in enumerate(tree.get_children(), start=1):
row_values = tree.item(item, 'values')
is_special = 'special' in tree.item(item, 'tags')
row_style = style_special if is_special else style
row_style_text = style_special_text if is_special else style_text
for j, value in enumerate(row_values):
if j == 12:
sheet.write(i, j, str(value), row_style_text)
elif j == 11:
raw_date = raw_values[i-1][10] if i-1 < len(raw_values) else None
if isinstance(raw_date, datetime.date):
date_style = style_date_special if is_special else style_date
sheet.write(i, j, raw_date, date_style)
elif raw_date:
sheet.write(i, j, str(raw_date), row_style)
else:
sheet.write(i, j, "", row_style)
else:
sheet.write(i, j, str(value), row_style)
workbook.save(save_path)
messagebox.showinfo("完成", f"数据成功导出至\n{save_path}")
except Exception as e:
messagebox.showerror("错误", f"导出失败：{str(e)}")
def copy_total_amount_to_clipboard():
root.clipboard_clear()
root.clipboard_append(f"不含税合计: {total_amount_sum:.2f} 元\n税额合计: {total_tax_sum:.2f} 元\n价税合计: {total_sum:.2f} 元")
messagebox.showinfo("成功", f"总金额统计已复制到剪贴板")
tree_frame = ttk.Frame(main_frame)
tree_frame.grid(column=0, row=0, pady=5, padx=5, sticky=(tk.N, tk.S, tk.E, tk.W))
tree_frame.columnconfigure(0, weight=1)
tree_frame.rowconfigure(0, weight=1)
scrollbar_y = ttk.Scrollbar(tree_frame, orient=tk.VERTICAL)
scrollbar_x = ttk.Scrollbar(tree_frame, orient=tk.HORIZONTAL)
tree = ttk.Treeview(tree_frame,
columns=("序号", "金额", "税率", "税额", "价税合计", "发票号码", "销售方名称", "销售方税号", "购买方名称", "购买方税号", "项目名称", "开票日期", "文件路径"),
show="headings",
yscrollcommand=scrollbar_y.set,
xscrollcommand=scrollbar_x.set)
scrollbar_y.config(command=tree.yview)
scrollbar_x.config(command=tree.xview)
tree.grid(column=0, row=0, sticky=(tk.N, tk.S, tk.E, tk.W))
scrollbar_y.grid(column=1, row=0, sticky=(tk.N, tk.S))
scrollbar_x.grid(column=0, row=1, sticky=(tk.E, tk.W))
tree.bind('<Double-1>', lambda event: open_pdf(tree.item(tree.selection())['values'][12]))
columns_config = {
"序号": {"width": 60, "anchor": "center", "stretch": False},
"金额": {"width": 120, "anchor": "center", "stretch": False},
"税率": {"width": 80, "anchor": "center", "stretch": False},
"税额": {"width": 120, "anchor": "center", "stretch": False},
"价税合计": {"width": 120, "anchor": "center", "stretch": False},
"发票号码": {"width": 280, "anchor": "center", "stretch": True},
"销售方名称": {"width": 400, "anchor": "w", "stretch": True},
"销售方税号": {"width": 200, "anchor": "center", "stretch": True},
"购买方名称": {"width": 260, "anchor": "w", "stretch": True},
"购买方税号": {"width": 200, "anchor": "center", "stretch": True},
"项目名称": {"width": 150, "anchor": "w", "stretch": False},
"开票日期": {"width": 120, "anchor": "center", "stretch": False},
"文件路径": {"width": 300, "anchor": "w", "stretch": True}
}
for col, config in columns_config.items():
tree.heading(col, text=col, command=lambda c=col: column_sorter(tree, c, 'num' if c in ['金额', '税率', '税额', '价税合计'] else 'str'))
tree.column(col, width=config["width"], anchor=config["anchor"], stretch=config.get("stretch", False))
tree.tag_configure('special', background='#FFFF99')
for index, value in enumerate(values, start=1):
tax_id = str(value[6]) if value[6] else ""
special_tax_ids = ["", "未识别", "航空客票无税号", "航空客票行程单无税号", "铁路客票无税号", "非税收入无税号", "行程单无税号", "无税号", "读取失败"]
has_no_valid_tax_id = tax_id in special_tax_ids
tags = ('special',) if has_no_valid_tax_id else ()
date_value = value[10]
if isinstance(date_value, datetime.date):
date_str = f"{date_value.year}年{date_value.month}月{date_value.day}日"
elif date_value is None:
date_str = ""
else:
date_str = str(date_value)
tree.insert("", "end", values=(
index,
f"{value[0]:.2f}" if value[0] > 0 else "0.00",
f"{round(value[1]):.0f}%" if value[1] > 0 else "0%",
f"{value[2]:.2f}" if value[2] > 0 else "0.00",
f"{value[3]:.2f}" if value[3] > 0 else "0.00",
value[4],
value[5],
value[6],
value[7] if value[7] else "",
value[8] if value[8] else "",
value[9],
date_str,
value[11]
), tags=tags)
button_frame = ttk.Frame(main_frame)
button_frame.grid(column=1, row=0, padx=10, sticky=(tk.N, tk.S))
buttons = [
("重命名选中文件", rename_selected_files),
("复制统计金额", copy_total_amount_to_clipboard),
("导出到XLS", export_to_xls),
("退出", root.destroy)
]
for i, (text, command) in enumerate(buttons):
btn = ttk.Button(button_frame, text=text, command=command, width=18)
btn.grid(column=0, row=i, pady=5)
info_frame = ttk.Frame(main_frame)
info_frame.grid(column=0, row=1, columnspan=2, pady=10, sticky=(tk.W, tk.E))
info_frame.columnconfigure(0, weight=1)
total_label = ttk.Label(info_frame, text=f"不含税合计: {total_amount_sum:.2f} 元 | 税额合计: {total_tax_sum:.2f} 元 | 价税合计: {total_sum:.2f} 元",
font=('Arial', 11, 'bold'), foreground='green')
total_label.grid(column=0, row=0, sticky=tk.W, padx=5)
stats_label = ttk.Label(info_frame, text=f"共处理 {len(values)} 张发票 | 双击行可打开PDF文件 | 黄色底色=非标准发票(航空/铁路/专利/行程单)", font=('Arial', 9), foreground='gray')
stats_label.grid(column=0, row=1, sticky=tk.W, padx=5)
root.mainloop()
def browse_folder(entry):
folder = filedialog.askdirectory(title="请选择发票PDF文件夹路径")
if folder:
entry.delete(0, tk.END)
entry.insert(0, folder)
def start_processing(entry, input_root):
logging.info('开始处理')
folder = entry.get()
if not folder:
messagebox.showerror("错误", "请先选择或输入一个文件夹路径")
return
global pdf_files_folder
pdf_files = get_pdf_files(folder)
if pdf_files:
pdf_files_folder = os.path.dirname(pdf_files[0])
values = []
error_files = []
for i, pdf_file in enumerate(pdf_files, 1):
logging.info(f'处理文件 {i}/{len(pdf_files)}: {os.path.basename(pdf_file)}')
pdf_content = read_pdf_content(pdf_file)
total_amount = pdf_content["total_amount"]
if total_amount == 0 and pdf_content["amount"] > 0:
total_amount = pdf_content["amount"]
if total_amount > 0:
display_project_name = pdf_content["project_name"] or pdf_content["category"]
values.append((
pdf_content["amount"],
pdf_content["tax_rate"],
pdf_content["tax_amount"],
pdf_content["total_amount"],
pdf_content["invoice_number"],
pdf_content["name"],
pdf_content["tax_id"],
pdf_content.get("buyer_name", ""),
pdf_content.get("buyer_tax_id", ""),
display_project_name,
pdf_content["date"],
pdf_file
))
else:
error_files.append(pdf_file)
logging.warning(f'无法提取有效金额: {pdf_file}')
if not values:
messagebox.showerror("错误", "没有找到有效的发票数据！\n请检查PDF文件是否为有效的电子发票。")
return
amounts = [value[0] for value in values]
tax_amounts = [value[2] for value in values]
total_amounts = [value[3] for value in values]
total_amount_sum = sum(amounts)
total_tax_sum = sum(tax_amounts)
total_sum = sum(total_amounts)
logging.info(f'不含税金额合计: {total_amount_sum}, 税额合计: {total_tax_sum}, 价税合计: {total_sum}, 有效文件数: {len(values)}')
if error_files:
messagebox.showwarning("警告", f"有 {len(error_files)} 个文件未能正确提取数据，已跳过。\n详细信息请查看 app.log 文件。")
display_results(values, total_amount_sum, total_tax_sum, total_sum, input_root, folder)
def main():
logging.info('程序启动')
root = tk.Tk()
root.title("发票金额统计系统")
center_window(root, width=700, height=300)
root.minsize(600, 280)
root.columnconfigure(0, weight=1)
root.rowconfigure(0, weight=1)
main_frame = ttk.Frame(root, padding="30")
main_frame.grid(column=0, row=0, sticky=(tk.W, tk.E, tk.N, tk.S))
main_frame.columnconfigure(0, weight=1)
title_label = ttk.Label(main_frame, text="发票金额统计工具", font=('Arial', 18, 'bold'))
title_label.grid(column=0, row=0, pady=(0, 20))
desc_label = ttk.Label(main_frame, text="自动提取电子发票中的金额、税额、价税合计、销售方信息及纳税人识别号", font=('Arial', 9), foreground='gray')
desc_label.grid(column=0, row=1, pady=(0, 20))
folder_frame = ttk.Frame(main_frame)
folder_frame.grid(column=0, row=2, sticky=(tk.W, tk.E), pady=5)
folder_frame.columnconfigure(0, weight=1)
folder_label = ttk.Label(folder_frame, text="选择发票文件夹:", font=('Arial', 10))
folder_label.grid(column=0, row=0, sticky=tk.W)
folder_entry = ttk.Entry(folder_frame, font=('Arial', 10))
folder_entry.grid(column=0, row=1, sticky=(tk.W, tk.E), pady=5)
browse_button = ttk.Button(folder_frame, text="浏览", command=lambda: browse_folder(folder_entry), width=10)
browse_button.grid(column=1, row=1, padx=(10, 0))
start_button = ttk.Button(main_frame, text="开始处理", command=lambda: start_processing(folder_entry, root), width=20)
start_button.grid(column=0, row=3, pady=20)
root.mainloop()
if __name__ == "__main__":
main()
[复制链接]
发表于 2026-6-11 20:50
|
发表于 2026-6-12 08:03
