怎样实现发票数据导出脚本

wen 实用脚本 2026-06-28 197

本文目录导读：

怎样实现发票数据导出脚本

Python版本（推荐，支持多种格式）
Node.js版本
SQL导出脚本（从数据库）
Shell脚本版本（Linux/Unix）
Web API导出端点示例（Python Flask）
安装依赖

我来帮您设计一个发票数据导出脚本,这里提供几种常见格式的解决方案：

Python版本（推荐，支持多种格式）

import pandas as pd
import csv
import json
from datetime import datetime
import os
class InvoiceExporter:
    """发票数据导出器"""
    def __init__(self, data_source):
        """
        初始化
        :param data_source: 数据源（可以是数据库连接、文件路径等）
        """
        self.data_source = data_source
        self.invoices = []
    def load_invoices(self):
        """加载发票数据（示例：从CSV文件）"""
        try:
            # 示例：从CSV加载数据
            if isinstance(self.data_source, str):
                self.invoices = pd.read_csv(self.data_source).to_dict('records')
            # 如果是数据库，可以：
            # self.invoices = db_query("SELECT * FROM invoices")
            return True
        except Exception as e:
            print(f"加载数据失败: {e}")
            return False
    def export_to_csv(self, output_file):
        """导出为CSV格式"""
        if not self.invoices:
            print("没有数据可导出")
            return False
        try:
            df = pd.DataFrame(self.invoices)
            df.to_csv(output_file, index=False, encoding='utf-8-sig')
            print(f"CSV文件已导出: {output_file}")
            return True
        except Exception as e:
            print(f"CSV导出失败: {e}")
            return False
    def export_to_excel(self, output_file):
        """导出为Excel格式"""
        if not self.invoices:
            print("没有数据可导出")
            return False
        try:
            df = pd.DataFrame(self.invoices)
            with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
                df.to_excel(writer, sheet_name='发票数据', index=False)
                # 自动调整列宽
                for column in df:
                    column_width = max(df[column].astype(str).str.len().max(), 
                                     len(column))
                    col_idx = df.columns.get_loc(column)
                    writer.sheets['发票数据'].column_dimensions[
                        chr(65 + col_idx)].width = min(column_width + 2, 50)
            print(f"Excel文件已导出: {output_file}")
            return True
        except Exception as e:
            print(f"Excel导出失败: {e}")
            return False
    def export_to_json(self, output_file):
        """导出为JSON格式"""
        if not self.invoices:
            print("没有数据可导出")
            return False
        try:
            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump(self.invoices, f, ensure_ascii=False, indent=2)
            print(f"JSON文件已导出: {output_file}")
            return True
        except Exception as e:
            print(f"JSON导出失败: {e}")
            return False
    def export_with_filter(self, output_file, format_type='csv', 
                          date_from=None, date_to=None, 
                          amount_min=None, amount_max=None):
        """带筛选条件的导出"""
        filtered = self.invoices.copy()
        # 日期筛选
        if date_from or date_to:
            if 'date' in filtered[0]:
                if date_from:
                    filtered = [i for i in filtered 
                               if i['date'] >= date_from]
                if date_to:
                    filtered = [i for i in filtered 
                               if i['date'] <= date_to]
        # 金额筛选
        if 'amount' in filtered[0]:
            if amount_min:
                filtered = [i for i in filtered 
                           if i['amount'] >= amount_min]
            if amount_max:
                filtered = [i for i in filtered 
                           if i['amount'] <= amount_max]
        # 临时替换数据
        original_data = self.invoices
        self.invoices = filtered
        # 根据格式导出
        if format_type == 'csv':
            result = self.export_to_csv(output_file)
        elif format_type == 'excel':
            result = self.export_to_excel(output_file)
        elif format_type == 'json':
            result = self.export_to_json(output_file)
        else:
            result = False
        # 恢复原始数据
        self.invoices = original_data
        return result
# 使用示例
if __name__ == "__main__":
    # 创建导出器实例
    exporter = InvoiceExporter("invoices.csv")  # 或数据库连接
    # 加载数据
    if exporter.load_invoices():
        # 导出为各种格式
        exporter.export_to_csv("导出发票.csv")
        exporter.export_to_excel("导出发票.xlsx")
        exporter.export_to_json("导出发票.json")
        # 带筛选的导出
        exporter.export_with_filter(
            "筛选发票.csv",
            format_type='csv',
            date_from="2024-01-01",
            date_to="2024-12-31",
            amount_min=1000
        )

Node.js版本

const fs = require('fs');
const path = require('path');
const ExcelJS = require('exceljs');
class InvoiceExporter {
    constructor(dataSource) {
        this.dataSource = dataSource;
        this.invoices = [];
    }
    async loadInvoices() {
        try {
            // 从文件或数据库加载数据
            const data = fs.readFileSync(this.dataSource, 'utf-8');
            this.invoices = JSON.parse(data);
            return true;
        } catch (error) {
            console.error('加载数据失败:', error);
            return false;
        }
    }
    async exportToExcel(outputFile) {
        const workbook = new ExcelJS.Workbook();
        const worksheet = workbook.addWorksheet('发票数据');
        // 添加列头
        if (this.invoices.length > 0) {
            const columns = Object.keys(this.invoices[0]).map(key => ({
                header: key,
                key: key,
                width: 20
            }));
            worksheet.columns = columns;
            // 添加数据
            this.invoices.forEach(invoice => {
                worksheet.addRow(invoice);
            });
            // 设置样式
            worksheet.getRow(1).font = { bold: true };
        }
        await workbook.xlsx.writeFile(outputFile);
        console.log(`Excel文件已导出: ${outputFile}`);
    }
    async exportToCSV(outputFile) {
        const headers = Object.keys(this.invoices[0]).join(',');
        const rows = this.invoices.map(invoice => 
            Object.values(invoice).map(value => 
                typeof value === 'string' && value.includes(',') ? 
                `"${value}"` : value
            ).join(',')
        );
        const csvContent = [headers, ...rows].join('\n');
        fs.writeFileSync(outputFile, '\uFEFF' + csvContent, 'utf-8');
        console.log(`CSV文件已导出: ${outputFile}`);
    }
}
// 使用示例
async function main() {
    const exporter = new InvoiceExporter('./invoices.json');
    await exporter.loadInvoices();
    await exporter.exportToExcel('./exported_invoices.xlsx');
    await exporter.exportToCSV('./exported_invoices.csv');
}
main().catch(console.error);

SQL导出脚本（从数据库）

-- MySQL/MariaDB
SELECT 
    invoice_id,
    invoice_number,
    customer_name,
    invoice_date,
    amount,
    tax,
    total_amount,
    status,
    created_at
INTO OUTFILE '/tmp/invoices_export.csv'
FIELDS TERMINATED BY ',' 
ENCLOSED BY '"'
LINES TERMINATED BY '\n'
FROM invoices
WHERE invoice_date BETWEEN '2024-01-01' AND '2024-12-31';
-- PostgreSQL
COPY (
    SELECT 
        invoice_id,
        invoice_number,
        customer_name,
        invoice_date,
        amount,
        tax,
        total_amount,
        status
    FROM invoices
    WHERE invoice_date BETWEEN '2024-01-01' AND '2024-12-31'
) TO '/tmp/invoices_export.csv' WITH CSV HEADER;
-- SQL Server
BCP "SELECT * FROM invoices WHERE invoice_date BETWEEN '2024-01-01' AND '2024-12-31'" 
QUERYOUT "C:\exports\invoices_export.csv" -c -t"," -S server_name -U username -P password

Shell脚本版本（Linux/Unix）

#!/bin/bash
# 发票数据导出脚本
DB_HOST="localhost"
DB_USER="root"
DB_PASS="password"
DB_NAME="invoice_system"
EXPORT_DIR="/var/exports/invoices"
# 创建导出目录
mkdir -p $EXPORT_DIR
# 获取当前日期
CURRENT_DATE=$(date +%Y%m%d_%H%M%S)
# 导出为CSV
echo "开始导出发票数据..."
mysql -h $DB_HOST -u $DB_USER -p$DB_PASS $DB_NAME -e "
SELECT 
    invoice_id,
    invoice_number,
    customer_name,
    invoice_date,
    amount,
    tax,
    total_amount,
    status
FROM invoices
WHERE invoice_date >= DATE_SUB(CURRENT_DATE, INTERVAL 30 DAY)
" > "$EXPORT_DIR/invoices_$CURRENT_DATE.csv"
if [ $? -eq 0 ]; then
    echo "✅ 发票数据导出成功: $EXPORT_DIR/invoices_$CURRENT_DATE.csv"
    echo "导出记录数: $(wc -l < $EXPORT_DIR/invoices_$CURRENT_DATE.csv)"
    # 可选：压缩文件
    gzip "$EXPORT_DIR/invoices_$CURRENT_DATE.csv"
    echo "文件已压缩"
else
    echo "❌ 导出失败"
    exit 1
fi

Web API导出端点示例（Python Flask）

from flask import Flask, request, jsonify, send_file
import pandas as pd
from io import BytesIO
import tempfile
app = Flask(__name__)
@app.route('/api/invoices/export', methods=['GET'])
def export_invoices():
    """发票导出API"""
    try:
        # 获取参数
        format_type = request.args.get('format', 'csv')
        date_from = request.args.get('from')
        date_to = request.args.get('to')
        # 查询数据（示例）
        # invoices = query_db("SELECT * FROM invoices WHERE ...")
        invoices = [
            {'id': 1, 'number': 'INV-001', 'amount': 1000, 'date': '2024-01-01'},
            {'id': 2, 'number': 'INV-002', 'amount': 2000, 'date': '2024-01-02'},
        ]
        df = pd.DataFrame(invoices)
        # 根据格式返回
        if format_type == 'csv':
            output = BytesIO()
            df.to_csv(output, index=False, encoding='utf-8-sig')
            output.seek(0)
            return send_file(
                output,
                mimetype='text/csv',
                as_attachment=True,
                download_name='invoices.csv'
            )
        elif format_type == 'excel':
            output = BytesIO()
            with pd.ExcelWriter(output, engine='openpyxl') as writer:
                df.to_excel(writer, index=False, sheet_name='发票')
            output.seek(0)
            return send_file(
                output,
                mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
                as_attachment=True,
                download_name='invoices.xlsx'
            )
        elif format_type == 'json':
            return jsonify(invoices)
        else:
            return jsonify({'error': '不支持的格式'}), 400
    except Exception as e:
        return jsonify({'error': str(e)}), 500
if __name__ == '__main__':
    app.run(debug=True)

安装依赖

# Python
pip install pandas openpyxl
# Node.js
npm install exceljs
# 或全局安装
npm install -g exceljs

选择适合您需求的版本进行修改和使用,建议根据实际的数据源（文件、数据库、API）和输出格式要求进行调整。

本文地址： https://www.bmjup.com/post/11031.html