def check_quality(image_path):
    img = cv2.imread(image_path)
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    
    brightness = np.mean(gray)
    sharpness = cv2.Laplacian(gray, cv2.CV_64F).var()
    
    return (80 < brightness < 200) and (sharpness > 100)

def preprocess(image_path):
    img = cv2.imread(image_path)
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    denoised = cv2.fastNlMeansDenoising(gray, None, 30, 7, 21)
    clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8))
    enhanced = clahe.apply(denoised)
    cv2.imwrite('processed.jpg', enhanced)

# 每批 50 张
for i in range(0, total, 50):
    batch = files[i:i+50]
    process_batch(batch)
    time.sleep(2)  # 批次间休息

def recognize_with_retry(file_path, max_retries=3):
    for i in range(max_retries):
        try:
            return recognize(file_path)
        except:
            if i < max_retries - 1:
                time.sleep(5)
            else:
                raise

def validate(data):
    if not re.match(r'^\d{10}|\d{12}$', data['发票代码']):
        return False
    if not re.match(r'^\d{8}$', data['发票号码']):
        return False
    if float(data['金额']) <= 0:
        return False
    return True

# 1. 检查文件大小
print(os.path.getsize(image_path))

# 2. 检查格式
from PIL import Image
img = Image.open(image_path)
print(img.format)

# 3. 检查尺寸
print(img.size)

# 打印请求参数
print(request_params)

# 检查必填字段
required = ['input_path', 'output_path', 'id', 'key']
for param in required:
    assert param in params, f"{param} is required"

# 检查并发数
print(concurrent_requests)

# 检查 API 端点
print(api_endpoint)
# 应该是：ocr.tencentcloudapi.com

# 1. 检查环境变量
import os
print(os.getenv('TENCENT_SECRET_ID'))

# 2. 重新获取密钥
# 访问：https://curl.qcloud.com/9ExTmaya

# 3. 更新配置
os.environ['TENCENT_SECRET_ID'] = '新密钥'

# 添加重试机制
def recognize_with_retry(file_path, max_retries=3):
    for i in range(max_retries):
        try:
            return poocr.ocr2excel.VatInvoiceOCR2Excel(...)
        except TimeoutError:
            if i < max_retries - 1:
                time.sleep(5)
            else:
                raise

# 检查并转换格式
from PIL import Image

def convert_image(image_path):
    img = Image.open(image_path)
    if img.format not in ['JPEG', 'PNG', 'PDF']:
        img = img.convert('RGB')
        img.save(image_path + '.jpg', 'JPEG')
    return image_path + '.jpg'

# 压缩图片
from PIL import Image

def compress_image(image_path, max_size=5):
    img = Image.open(image_path)
    
    # 检查文件大小
    file_size = os.path.getsize(image_path) / 1024 / 1024
    
    if file_size > max_size:
        # 压缩
        img.save(image_path, quality=80, optimize=True)
    
    return image_path

# 添加质量检查
def check_image_quality(image_path):
    img = cv2.imread(image_path)
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    
    # 检查亮度
    if np.mean(gray) < 80:
        return False, "图片太暗"
    
    # 检查清晰度
    if cv2.Laplacian(gray, cv2.CV_64F).var() < 100:
        return False, "图片模糊"
    
    return True, "合格"

# 安全获取字段
invoice_code = invoice_data.get('发票代码', '')
if not invoice_code:
    print("⚠️ 发票代码识别失败，需要人工复核")

# 检查文件是否被占用
def is_file_in_use(file_path):
    try:
        with open(file_path, 'r+') as f:
            pass
        return False
    except PermissionError:
        return True

# 等待文件释放
while is_file_in_use(output_path):
    time.sleep(1)

# 指定编码
with open(file_path, 'r', encoding='utf-8') as f:
    content = f.read()

# 或尝试多种编码
for encoding in ['utf-8', 'gbk', 'gb2312']:
    try:
        with open(file_path, 'r', encoding=encoding) as f:
            content = f.read()
        break
    except:
        continue

# 使用绝对路径
import os
abs_path = os.path.abspath(file_path)

# 检查路径是否存在
if not os.path.exists(abs_path):
    print(f"❌ 文件不存在：{abs_path}")

# 检查剩余额度
def check_quota():
    response = requests.get(
        'https://console.cloud.tencent.com/ocr/quota',
        headers={'Authorization': get_auth_token()}
    )
    quota = response.json()
    print(f"剩余额度：{quota['remaining']}")
    
    if quota['remaining'] < 100:
        print("⚠️ 额度不足，请充值")

def robust_invoice_recognition(file_path):
    """健壮的发票识别"""
    try:
        # 1. 检查文件
        if not os.path.exists(file_path):
            return None, "文件不存在"
        
        # 2. 检查质量
        is_ok, msg = check_image_quality(file_path)
        if not is_ok:
            return None, msg
        
        # 3. 识别
        result = poocr.ocr2excel.VatInvoiceOCR2Excel(...)
        
        # 4. 验证结果
        if not result.get('发票代码'):
            return None, "识别失败"
        
        return result, "成功"
        
    except Exception as e:
        return None, f"异常：{str(e)}"

import poocr
import requests
import hashlib

# 配置
SECRET_ID = os.getenv('TENCENT_SECRET_ID')
SECRET_KEY = os.getenv('TENCENT_SECRET_KEY')
TAX_API_URL = "https://inv-veri.chinatax.gov.cn/api"

def recognize_invoice(image_path):
    """识别发票"""
    result = poocr.ocr2excel.VatInvoiceOCR2Excel(
        input_path=image_path,
        output_path='./temp',
        id=SECRET_ID,
        key=SECRET_KEY
    )
    return parse_result(result)

def verify_invoice(invoice_data):
    """验真发票"""
    # 准备验真参数
    params = {
        'invoice_code': invoice_data['发票代码'],
        'invoice_number': invoice_data['发票号码'],
        'invoice_date': invoice_data['开票日期'],
        'amount': invoice_data['金额']
    }
    
    # 调用税务验真 API
    response = requests.post(
        f"{TAX_API_URL}/verify",
        json=params,
        headers={'Authorization': get_auth_token()}
    )
    
    result = response.json()
    
    if result['status'] == 'valid':
        return True, "验真通过"
    elif result['status'] == 'invalid':
        return False, "发票不存在或信息不符"
    elif result['status'] == 'cancelled':
        return False, "发票已作废"
    else:
        return None, "验真失败"

def process_invoice_with_verify(image_path):
    """识别 + 验真完整流程"""
    print(f"📄 处理发票：{image_path}")
    
    # 1. OCR 识别
    print("🔍 正在识别...")
    invoice_data = recognize_invoice(image_path)
    print(f"✅ 识别完成：{invoice_data['发票号码']}")
    
    # 2. 税务验真
    print("🔐 正在验真...")
    is_valid, message = verify_invoice(invoice_data)
    
    if is_valid:
        print(f"✅ 验真通过：{message}")
        save_to_database(invoice_data, status='valid')
    elif is_valid is False:
        print(f"❌ 验真失败：{message}")
        save_to_database(invoice_data, status='invalid')
    else:
        print(f"⚠️ 验真异常：{message}")
        save_to_database(invoice_data, status='pending')
    
    return invoice_data, is_valid, message

# 批量处理
for invoice_file in os.listdir('./invoices'):
    process_invoice_with_verify(f'./invoices/{invoice_file}')

import poocr

def recognize_invoice(file_path):
    """识别单张发票"""
    result = poocr.ocr2excel.VatInvoiceOCR2Excel(
        input_path=file_path,
        output_path='./temp',
        id=SECRET_ID,
        key=SECRET_KEY
    )
    return parse_result(result)

import sqlite3

def create_database():
    """创建数据库"""
    conn = sqlite3.connect('invoices.db')
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS invoices (
            id INTEGER PRIMARY KEY,
            invoice_code TEXT,
            invoice_number TEXT,
            date TEXT,
            amount REAL,
            tax REAL,
            total REAL,
            status TEXT,
            created_at TIMESTAMP
        )
    ''')
    
    conn.commit()
    conn.close()

import streamlit as st

st.title("发票管理系统")

uploaded_file = st.file_uploader("上传发票", type=['pdf', 'jpg', 'png'])

if uploaded_file:
    # 保存文件
    with open(f"./uploads/{uploaded_file.name}", "wb") as f:
        f.write(uploaded_file.getbuffer())
    
    # 识别
    result = recognize_invoice(f"./uploads/{uploaded_file.name}")
    
    # 显示结果
    st.success("识别成功！")
    st.json(result)
    
    # 保存到数据库
    save_to_database(result)

def get_statistics():
    """获取统计数据"""
    conn = sqlite3.connect('invoices.db')
    
    query = '''
        SELECT 
            COUNT(*) as total_count,
            SUM(amount) as total_amount,
            SUM(tax) as total_tax,
            SUM(total) as grand_total
        FROM invoices
        WHERE date BETWEEN ? AND ?
    '''
    
    result = pd.read_sql(query, conn, params=[start_date, end_date])
    return result

import pandas as pd

# 读取识别结果
df = pd.read_excel("发票识别结果.xlsx")

# 重命名列
df.rename(columns={
    '开票日期': '单据日期',
    '发票号码': '发票号',
    '不含税金额': '金额'
}, inplace=True)

# 选择需要的列
df = df[['单据日期', '发票号', '金额', '税额', '价税合计']]

# 保存
df.to_excel("用友 U8 导入模板.xlsx", index=False)

import pandas as pd

# 读取识别结果
df_source = pd.read_excel("发票识别结果.xlsx")

# 读取金蝶模板
df_template = pd.read_excel("金蝶导入模板.xlsx")

# 数据映射
mapping = {
    '开票日期': 'FDate',
    '发票号码': 'FInvoiceNo',
    '金额': 'FAmount',
    '税额': 'FTax',
    '价税合计': 'FTotalAmount'
}

for source_col, target_col in mapping.items():
    df_template[target_col] = df_source[source_col]

# 保存
df_template.to_excel("金蝶导入数据.xlsx", index=False)

import pandas as pd

df = pd.read_excel("发票识别结果.xlsx")

# SAP 字段要求
sap_df = pd.DataFrame()
sap_df['BKPF-BUDAT'] = df['开票日期']  # 过账日期
sap_df['BKPF-BLART'] = 'RE'  # 凭证类型
sap_df['BKPF-BUKRS'] = '1000'  # 公司代码
sap_df['BSEG-WRBTR'] = df['价税合计']  # 金额
sap_df['BSEG-MWSKZ'] = 'V1'  # 税码

# 保存为 CSV（UTF-8，分号分隔）
sap_df.to_csv("SAP_导入.csv", sep=';', index=False, encoding='utf-8')

import pandas as pd
import os

def auto_import_to_erp(invoice_folder, erp_system='yonyou'):
    """自动导入 ERP 系统"""
    
    # 1. 识别发票
    poocr.ocr2excel.VatInvoiceOCR2Excel(
        input_path=invoice_folder,
        output_path='./output',
        id=SECRET_ID,
        key=SECRET_KEY
    )
    
    # 2. 读取识别结果
    df = pd.read_excel("./output/发票识别结果.xlsx")
    
    # 3. 根据 ERP 系统转换格式
    if erp_system == 'yonyou':
        df = convert_to_yonyou(df)
        output_file = "用友导入模板.xlsx"
    elif erp_system == 'kingdee':
        df = convert_to_kingdee(df)
        output_file = "金蝶导入模板.xlsx"
    elif erp_system == 'sap':
        df = convert_to_sap(df)
        output_file = "SAP_导入.csv"
    
    # 4. 保存
    df.to_excel(f"./output/{output_file}", index=False)
    
    print(f"✅ 导入文件已生成：./output/{output_file}")
    print("请手动导入到 ERP 系统")

print("Hello, 发票！")

invoice_code = "011002100113"  # 发票代码
amount = 1000.00  # 金额
date = "2026-03-17"  # 日期

print(f"发票 {invoice_code} 金额 {amount}")

invoices = ["发票 1", "发票 2", "发票 3"]

for invoice in invoices:
    print(f"处理：{invoice}")

amount = 1000

if amount > 5000:
    print("大额发票，需要审核")
else:
    print("普通发票")

def process_invoice(code, amount):
    print(f"处理发票 {code}，金额 {amount}")

process_invoice("011002100113", 1000)

import os

files = os.listdir("./invoices")
print(f"找到 {len(files)} 个文件")

import os

for i, filename in enumerate(os.listdir("./invoices")):
    new_name = f"发票_{i+1}.pdf"
    os.rename(f"./invoices/{filename}", f"./invoices/{new_name}")

import pandas as pd

# 读取 Excel
df = pd.read_excel("invoices.xlsx")

# 查看数据
print(df.head())

# 写入 Excel
df.to_excel("output.xlsx", index=False)

import csv

with open("invoices.csv", "r", encoding="utf-8") as f:
    reader = csv.reader(f)
    for row in reader:
        print(row)

import pdfplumber

with pdfplumber.open("invoice.pdf") as pdf:
    for page in pdf.pages:
        text = page.extract_text()
        print(text)

import poocr

poocr.ocr2excel.VatInvoiceOCR2Excel(
    input_path="invoice.pdf",
    output_path="./output",
    id=SECRET_ID,
    key=SECRET_KEY
)

import os

for filename in os.listdir("./invoices"):
    poocr.ocr2excel.VatInvoiceOCR2Excel(
        input_path=f"./invoices/{filename}",
        output_path="./output",
        id=SECRET_ID,
        key=SECRET_KEY
    )

try:
    # 识别逻辑
except Exception as e:
    print(f"识别失败：{e}")

# 照抄的代码
poocr.ocr2excel.VatInvoiceOCR2Excel(
    input_path='C:/invoices',  # 改成自己的路径就报错
    output_path='./output',
    id='SecretId',
    key='SecretKey'
)

# input_path = 发票在哪里
# output_path = 结果放哪里
# id = 腾讯云账号 ID
# key = 腾讯云账号密钥

import cv2
import numpy as np

def preprocess_image(image_path, output_path):
    """图片预处理"""
    img = cv2.imread(image_path)
    
    # 1. 转为灰度图
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    
    # 2. 去噪
    denoised = cv2.fastNlMeansDenoising(gray, None, 30, 7, 21)
    
    # 3. 增强对比度
    clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8))
    enhanced = clahe.apply(denoised)
    
    # 4. 二值化（可选）
    _, binary = cv2.threshold(enhanced, 0, 255, 
                              cv2.THRESH_BINARY + cv2.THRESH_OTSU)
    
    # 保存
    cv2.imwrite(output_path, binary)
    print(f"✅ 图片预处理完成：{output_path}")

# 使用
preprocess_image('invoice.jpg', 'invoice_processed.jpg')

def check_image_quality(image_path):
    """检查图片质量"""
    img = cv2.imread(image_path)
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    
    checks = {}
    
    # 1. 检查亮度
    brightness = np.mean(gray)
    checks['brightness'] = brightness
    if brightness < 80:
        return False, f"图片太暗（亮度：{brightness:.1f}）"
    
    # 2. 检查清晰度
    variance = cv2.Laplacian(gray, cv2.CV_64F).var()
    checks['sharpness'] = variance
    if variance < 100:
        return False, f"图片模糊（清晰度：{variance:.1f}）"
    
    # 3. 检查尺寸
    height, width = gray.shape
    checks['size'] = (width, height)
    if width < 800 or height < 600:
        return False, f"图片太小（{width}x{height}）"
    
    return True, "合格"

# 使用
is_ok, msg = check_image_quality('invoice.jpg')
if not is_ok:
    print(f"❌ {msg}，请重新拍照")
else:
    print("✅ 图片质量合格")

def process_in_batches(files, batch_size=50):
    """分批处理"""
    total = len(files)
    success = 0
    failed = []
    
    for i in range(0, total, batch_size):
        batch = files[i:i+batch_size]
        print(f"📦 处理第{i//batch_size + 1}批（{len(batch)}张）")
        
        for file_path in batch:
            ok, error = recognize_with_retry(file_path)
            if ok:
                success += 1
            else:
                failed.append((file_path, error))
        
        # 每批之间休息 2 秒，避免限流
        time.sleep(2)
    
    return success, failed

def recognize_with_retry(file_path, max_retries=3):
    """带重试的识别"""
    for i in range(max_retries):
        try:
            # 识别逻辑
            return True, None
        except Exception as e:
            if i < max_retries - 1:
                time.sleep(5)  # 5 秒后重试
            else:
                return False, str(e)

import re

def validate_invoice_data(data):
    """校验发票数据"""
    errors = []
    
    # 1. 校验发票代码（10 位或 12 位数字）
    if not re.match(r'^\d{10}|\d{12}$', data['发票代码']):
        errors.append(f"发票代码格式错误：{data['发票代码']}")
    
    # 2. 校验发票号码（8 位数字）
    if not re.match(r'^\d{8}$', data['发票号码']):
        errors.append(f"发票号码格式错误：{data['发票号码']}")
    
    # 3. 校验金额（数字）
    try:
        amount = float(data['金额'])
        if amount < 0:
            errors.append(f"金额为负数：{amount}")
    except:
        errors.append(f"金额格式错误：{data['金额']}")
    
    # 4. 校验日期
    try:
        datetime.strptime(data['开票日期'], '%Y-%m-%d')
    except:
        errors.append(f"日期格式错误：{data['开票日期']}")
    
    return len(errors) == 0, errors

# 使用
is_valid, errors = validate_invoice_data(invoice_data)
if not is_valid:
    print(f"⚠️ 数据校验失败：{errors}")
    # 标记为需要人工复核

import json

def log_failed_invoice(filename, error, image_path):
    """记录失败发票"""
    log_entry = {
        'filename': filename,
        'error': error,
        'image_path': image_path,
        'time': datetime.now().isoformat(),
        'status': 'pending'  # pending/fixed/ignored
    }
    
    # 追加到日志文件
    log_file = './failed_invoices.json'
    if os.path.exists(log_file):
        with open(log_file, 'r', encoding='utf-8') as f:
            logs = json.load(f)
    else:
        logs = []
    
    logs.append(log_entry)
    
    with open(log_file, 'w', encoding='utf-8') as f:
        json.dump(logs, f, ensure_ascii=False, indent=2)

# 定期分析失败原因
def analyze_failures():
    """分析失败原因"""
    with open('./failed_invoices.json', 'r', encoding='utf-8') as f:
        logs = json.load(f)
    
    error_types = {}
    for log in logs:
        error_type = log['error'].split(':')[0]
        error_types[error_type] = error_types.get(error_type, 0) + 1
    
    print("失败原因统计：")
    for error_type, count in sorted(error_types.items(), key=lambda x: -x[1]):
        print(f"  {error_type}: {count}")