项目背景#

公司同时管理 3 家关联企业的财务，每月需要处理：

  • 销项发票：约 100 张，开具、登记、归档
  • 进项发票：约 200+ 张，查验、匹配、税务认证
  • 费用报销：多人多公司，按项目归集
  • 跨公司分析：管理层需要看 3 家公司的合并视图

原来的方案是 Excel + VBA + Streamlit 的拼凑架构，问题越来越多：

痛点 具体表现
数据孤岛 3 家公司各一套 Excel，很难合并看
权限缺失 所有人看所有数据，敏感信息无保护
无法追溯 数据被改了不知道谁改的、改了什么
手动调度 每月导出税局数据、跑脚本都要手动触发
扩展困难 加一个新功能就要改好几个 Excel 文件

目标：用一套 Django 系统统一管理，既是操作台（录入、审批），也是看板（多维分析、趋势对比）。

技术架构#

整体分层#

┌────────────────────────────────────────────┐
│         前端展示层                          │
│   Django Templates + ECharts + DataTables  │
├────────────────────────────────────────────┤
│         API 层                             │
│         Django REST Framework              │
├────────────────────────────────────────────┤
│         业务逻辑层                          │
│         Views / Services / Forms           │
├────────────────────────────────────────────┤
│         数据处理层                          │
│         Pandas / ETL Pipeline / Celery     │
├────────────────────────────────────────────┤
│         数据持久层                          │
│         Django ORM / MySQL                 │
├────────────────────────────────────────────┤
│         数据源                             │
│   税务局导出 / 云悦 CRM / Excel / API      │
└────────────────────────────────────────────┘

App 模块划分#

apps/
├── core/           # 基础：用户、权限、公共 BaseModel
├── invoices/       # 发票管理（进项/销项）
├── analytics/      # 数据分析与看板
├── etl/            # ETL 任务调度与日志
├── reports/        # 定时报表生成
├── inventory/      # 库存管理
├── expenses/       # 费用报销
└── crm/            # 云悦 CRM 数据同步

共设计 24 个数据库模型，跨 7 个业务模块。

技术选型说明#

  • Django ORM + MySQL：财务数据对事务完整性要求高，关系型数据库首选
  • Celery + Redis：税局数据同步、报表生成等耗时任务异步处理
  • Pandas：ETL 环节的数据清洗和转换，比纯 ORM 操作快得多
  • ECharts：前端图表，交互体验好，支持中文，文档完善
  • DRF：数据看板的 API 层，方便后续对接移动端

核心功能#

1. 发票全生命周期管理#

进项发票从税局导入到认证完成，每个环节都有状态跟踪：

税局文件导入 → 自动解析 → 数据校验 → 匹配采购合同 → 认证 → 归档

销项发票支持批量开票计划、开具记录登记和月度对账。

系统通过模糊匹配算法自动关联发票与采购合同，匹配率约 85%，剩余 15% 人工确认。

2. ETL 数据同步流水线#

税务局每月导出的 Excel 格式不固定，需要一套健壮的解析器：

class TaxBureauInvoiceImporter(BaseImporter):
    """税务局导出文件导入器"""

    COLUMN_MAPPING = {
        '发票号码': 'invoice_number',
        '开票日期': 'invoice_date',
        '购买方名称': 'buyer_name',
        '合计金额': 'amount_without_tax',
        '合计税额': 'tax_amount',
        '价税合计': 'total_amount',
    }

    def transform(self, df):
        # 日期列格式不固定，做容错处理
        df['invoice_date'] = pd.to_datetime(
            df['invoice_date'], errors='coerce'
        ).dt.date

        # 金额列可能含千分位逗号
        for col in ['amount_without_tax', 'tax_amount', 'total_amount']:
            df[col] = df[col].astype(str)\
                .str.replace(',', '')\
                .pipe(pd.to_numeric, errors='coerce')\
                .fillna(0)

        return df.dropna(subset=['invoice_number', 'invoice_date'])

每次导入都会生成任务日志，记录成功/失败条数和错误详情，方便排查。

3. 多维数据分析看板#

面向管理层提供以下视图：

  • 月度趋势：销项/进项金额按月展示，支持同比对比
  • 跨公司合并视图：3 家公司的数据在同一张图上对比
  • 税负分析：每月进销项税额差，支持按税率分组
  • Top 供应商 / 客户：按金额排序，可钻取到明细
  • 应付账款追踪：未付款发票汇总，按账期分层

图表均通过 API + ECharts 实现，支持时间范围筛选和公司切换。

4. 自动化调度#

通过 Celery Beat 实现定时任务：

app.conf.beat_schedule = {
    # 每天凌晨 2 点检查税局新数据
    'sync-tax-bureau': {
        'task': 'apps.etl.tasks.sync_invoices',
        'schedule': crontab(hour=2, minute=0),
    },
    # 每月 1 日早 8 点生成上月汇总报告
    'monthly-report': {
        'task': 'apps.reports.tasks.generate_monthly_report',
        'schedule': crontab(day_of_month=1, hour=8),
    },
}

报告自动发送到管理层邮件，不再需要人工触发。

数据模型设计亮点#

时间维度预计算#

发票模型在保存时自动填充 year、month、quarter 字段，避免分析查询时做日期函数运算，这对月度/季度聚合查询有显著性能提升：

def save(self, *args, **kwargs):
    if self.invoice_date:
        self.year = self.invoice_date.year
        self.month = self.invoice_date.month
        self.quarter = (self.invoice_date.month - 1) // 3 + 1
    super().save(*args, **kwargs)

分析专用 Manager#

复杂的聚合逻辑集中在自定义 Manager 中，View 层只调用语义清晰的方法：

# 调用方式
Invoice.analytics.monthly_summary(year=2024, direction='inbound')
Invoice.analytics.yoy_comparison(current_year=2024)
Invoice.analytics.top_suppliers(year=2024, top_n=10)

财务数据用 DecimalField#

所有金额字段一律用 DecimalField，不用 FloatField。浮点误差在财务对账场景下是不可接受的。

成果#

指标 优化前 优化后 变化
月度发票处理时长 约 25 小时 约 3 小时 -88%
数据错误率 ~5% <0.5% -90%
管理报告出具时间 月底次日 自动当天 提前 1 天
跨公司数据查询 手工汇总 Excel 秒级查询 质的飞跃
历史数据追溯 翻 Excel 文件 全文检索 —

当前进展与计划#

已完成：发票管理、ETL 流水线、多维分析看板、Celery 定时任务、基础权限体系

进行中：税务申报记录管理、费用报销审批流程、云悦 CRM 数据同步

计划中：移动端 API 适配、ClickHouse 分列式存储（数据量超千万时）

经验总结#

  1. 分层要彻底：Model 管数据结构，Manager 管查询逻辑，Service 管业务聚合，View 只管 HTTP——不要把聚合查询写在 View 里，后期很难维护

  2. ETL 必须有状态：每次数据导入都应该记录任务记录，出了问题能定位到具体哪批数据、哪条报错

  3. 缓存要主动失效：分析结果缓存在 Redis，但发票数据一旦变更，要通过 Django signal 精确清除对应缓存，否则数据会不一致

  4. 索引要和查询对齐：复合索引的字段顺序要和最常见的查询过滤条件保持一致，建模时就要想好，事后改代价很高

  5. 财务数据的 Decimal 原则：哪怕觉得精度够用，也永远用 DecimalField，不要因为"方便"用 float