351 lines
15 KiB
Python
351 lines
15 KiB
Python
"""SQLite 连接与幂等初始化（票据合规预审助手）。
|
||
|
||
业务表 5 张：
|
||
- invoice_records（发票登记，只读，预审自动写入）
|
||
- prereview_reports（初审意见，结论只读；查验回填走 mark-verified）
|
||
- expense_standards（经费标准库，可编辑）
|
||
- taxid_whitelist（税号白名单，可编辑）
|
||
- holiday_calendar（节假日，可编辑）
|
||
|
||
表间禁止 SQLite 外键，仅存业务 id（见 references/SCHEMA.md）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
|
||
from db.display_metadata import (
|
||
init_display_metadata,
|
||
seed_column_metadata_batch,
|
||
upsert_table_metadata,
|
||
)
|
||
from db.seed_data import seed_all
|
||
from db.timestamp_columns import init_updated_at_trigger
|
||
from util.runtime_paths import get_db_path
|
||
|
||
|
||
def get_conn() -> sqlite3.Connection:
|
||
return sqlite3.connect(get_db_path())
|
||
|
||
|
||
_INVOICE_RECORDS_DDL = """
|
||
CREATE TABLE IF NOT EXISTS invoice_records (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
batch_no TEXT NOT NULL,
|
||
invoice_type TEXT,
|
||
invoice_code TEXT,
|
||
invoice_number TEXT NOT NULL,
|
||
issue_date TEXT,
|
||
amount REAL,
|
||
amount_in_words TEXT,
|
||
tax_rate REAL,
|
||
buyer_name TEXT,
|
||
buyer_tax_no TEXT,
|
||
seller_name TEXT,
|
||
seller_tax_no TEXT,
|
||
expense_category TEXT,
|
||
city TEXT,
|
||
days INTEGER,
|
||
purpose TEXT,
|
||
project_no TEXT,
|
||
trip_context_json TEXT,
|
||
created_at INTEGER NOT NULL DEFAULT (unixepoch())
|
||
)
|
||
"""
|
||
|
||
_INVOICE_RECORDS_INDEX = """
|
||
CREATE INDEX IF NOT EXISTS idx_invoice_records_batch
|
||
ON invoice_records(batch_no)
|
||
"""
|
||
|
||
_PREREVIEW_REPORTS_DDL = """
|
||
CREATE TABLE IF NOT EXISTS prereview_reports (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
batch_no TEXT NOT NULL,
|
||
invoice_id INTEGER,
|
||
invoice_number TEXT,
|
||
seller_name TEXT,
|
||
amount REAL,
|
||
verdict TEXT NOT NULL,
|
||
pass_json TEXT,
|
||
warn_json TEXT,
|
||
reject_json TEXT,
|
||
summary TEXT,
|
||
verify_status TEXT NOT NULL DEFAULT '待查验',
|
||
verify_note TEXT,
|
||
verify_updated_at INTEGER,
|
||
report_file TEXT,
|
||
created_at INTEGER NOT NULL DEFAULT (unixepoch())
|
||
)
|
||
"""
|
||
|
||
_PREREVIEW_REPORTS_INDEX = """
|
||
CREATE INDEX IF NOT EXISTS idx_prereview_reports_batch
|
||
ON prereview_reports(batch_no)
|
||
"""
|
||
|
||
_EXPENSE_STANDARDS_DDL = """
|
||
CREATE TABLE IF NOT EXISTS expense_standards (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
category TEXT NOT NULL,
|
||
city_tier TEXT NOT NULL,
|
||
city_example TEXT,
|
||
staff_title TEXT NOT NULL,
|
||
max_amount REAL NOT NULL,
|
||
unit TEXT,
|
||
basis_doc TEXT,
|
||
clause TEXT,
|
||
effective_date INTEGER,
|
||
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
|
||
updated_at INTEGER NOT NULL DEFAULT (unixepoch()),
|
||
UNIQUE (category, city_tier, staff_title)
|
||
)
|
||
"""
|
||
|
||
_TAXID_WHITELIST_DDL = """
|
||
CREATE TABLE IF NOT EXISTS taxid_whitelist (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
tax_no TEXT NOT NULL UNIQUE,
|
||
org_name TEXT,
|
||
note TEXT,
|
||
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
|
||
updated_at INTEGER NOT NULL DEFAULT (unixepoch())
|
||
)
|
||
"""
|
||
|
||
_HOLIDAY_CALENDAR_DDL = """
|
||
CREATE TABLE IF NOT EXISTS holiday_calendar (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
holiday_date TEXT NOT NULL UNIQUE,
|
||
holiday_name TEXT,
|
||
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
|
||
updated_at INTEGER NOT NULL DEFAULT (unixepoch())
|
||
)
|
||
"""
|
||
|
||
_TS_COLUMN_NAMES = {"created_at", "updated_at", "effective_date", "verify_updated_at"}
|
||
|
||
|
||
def _col(column_name: str, display_name: str, description: str, **extra: object) -> dict[str, object]:
|
||
base: dict[str, object] = {
|
||
"column_name": column_name,
|
||
"display_name": display_name,
|
||
"description": description,
|
||
"visible": 1,
|
||
"searchable": 0,
|
||
"editable": 0,
|
||
}
|
||
base.update(extra)
|
||
if column_name in _TS_COLUMN_NAMES:
|
||
base["display_type"] = "datetime_unix_seconds"
|
||
return base
|
||
|
||
|
||
_INVOICE_RECORDS_COLUMNS: tuple[dict[str, object], ...] = (
|
||
_col("id", "编号", "主键", searchable=1),
|
||
_col("batch_no", "预审批次号", "同一次预审请求的批次号，同批多张发票共享。", searchable=1),
|
||
_col("invoice_type", "发票类型", "如增值税电子普通发票、增值税专用发票、出租车票等。", searchable=1),
|
||
_col("invoice_code", "发票代码", "票面发票代码（数电票可能为空）。", searchable=1),
|
||
_col("invoice_number", "发票号码", "票面发票号码。", searchable=1),
|
||
_col("issue_date", "开票日期", "票面开票日期（YYYY-MM-DD）。", searchable=1),
|
||
_col("amount", "金额", "票面价税合计或金额（元）。", searchable=1, display_type="number"),
|
||
_col("amount_in_words", "大写金额", "票面大写金额（默认隐藏）。", visible=0),
|
||
_col("tax_rate", "税率", "票面税率（小数，如 0.06 表示 6%）。", display_type="number"),
|
||
_col("buyer_name", "购买方名称", "票面购买方全称。", searchable=1),
|
||
_col("buyer_tax_no", "购买方税号", "购买方统一社会信用代码，用于白名单比对。", searchable=1),
|
||
_col("seller_name", "销售方名称", "票面销售方全称。", searchable=1),
|
||
_col("seller_tax_no", "销售方税号", "销售方统一社会信用代码，用于集中开票风控。", searchable=1),
|
||
_col("expense_category", "费用类别", "报销费用类别，如住宿费、交通费、会议费。", searchable=1),
|
||
_col("city", "消费城市", "发票对应消费城市（住宿/交通判断用）。", searchable=1),
|
||
_col("days", "天数", "该发票覆盖的天数（住宿费核算用，缺省按 1 天）。", searchable=1, display_type="number"),
|
||
_col("purpose", "报销事由", "用户填报的报销事由文字。", searchable=1),
|
||
_col("project_no", "预算项目编号", "系统带入的预算项目编号。", searchable=1),
|
||
_col("trip_context_json", "差旅上下文", "出差起止、目的地、职级等上下文（JSON）。", visible=0, display_type="textarea"),
|
||
_col("created_at", "创建时间", "记录入库时间（Unix 秒级时间戳，自动生成）。"),
|
||
)
|
||
|
||
_PREREVIEW_REPORTS_COLUMNS: tuple[dict[str, object], ...] = (
|
||
_col("id", "报告编号", "主键，初审意见单编号。", searchable=1),
|
||
_col("batch_no", "预审批次号", "同一次预审请求的批次号。", searchable=1),
|
||
_col("invoice_id", "发票编号", "关联发票登记表主键。", searchable=1),
|
||
_col("invoice_number", "发票号码", "本报告对应的发票号码。", searchable=1),
|
||
_col("seller_name", "销售方名称", "本报告对应的销售方。", searchable=1),
|
||
_col("amount", "金额", "本报告对应的票面金额（元）。", searchable=1, display_type="number"),
|
||
_col(
|
||
"verdict",
|
||
"预审结论",
|
||
"通过 / 待确认 / 驳回。",
|
||
searchable=1,
|
||
options_json='[{"value":"通过","label":"✅ 通过"},{"value":"待确认","label":"⚠️ 待确认"},{"value":"驳回","label":"❌ 驳回"}]',
|
||
),
|
||
_col("pass_json", "通过项", "通过项清单（JSON 数组）。", visible=0, display_type="textarea"),
|
||
_col("warn_json", "待确认项", "待确认项清单（JSON 数组，附规则引用）。", visible=0, display_type="textarea"),
|
||
_col("reject_json", "驳回项", "驳回项清单（JSON 数组，附违反条款）。", visible=0, display_type="textarea"),
|
||
_col("summary", "结论摘要", "结论一句话摘要。", searchable=1),
|
||
_col(
|
||
"verify_status",
|
||
"真伪查验",
|
||
"待查验 / 已查验通过 / 已查验不符。",
|
||
searchable=1,
|
||
options_json='[{"value":"待查验","label":"待查验"},{"value":"已查验通过","label":"已查验通过"},{"value":"已查验不符","label":"已查验不符"}]',
|
||
),
|
||
_col("verify_note", "查验备注", "人工查验回填的备注。", searchable=1),
|
||
_col("verify_updated_at", "查验回填时间", "最近一次人工查验回填时间（Unix 秒级）。"),
|
||
_col("report_file", "意见单文件", "《初审意见单》HTML 文件路径。", visible=0),
|
||
_col("created_at", "创建时间", "报告创建时间（Unix 秒级时间戳，自动生成）。"),
|
||
)
|
||
|
||
_EXPENSE_STANDARDS_COLUMNS: tuple[dict[str, object], ...] = (
|
||
_col("id", "编号", "主键", searchable=1),
|
||
_col("category", "费用类别", "住宿费 / 交通费等。", searchable=1),
|
||
_col("city_tier", "城市等级", "一线城市 / 省会及计划单列市 / 其他地区 / 通用。", searchable=1),
|
||
_col("city_example", "城市示例", "该等级覆盖的城市（逗号分隔），用于城市归档。", searchable=1, editable=1),
|
||
_col("staff_title", "职级", "如 副高及以上 / 中级及以下。", searchable=1),
|
||
_col("max_amount", "标准值", "住宿费为上限金额（元）；交通费为席位等级代码。", searchable=1, editable=1, display_type="number"),
|
||
_col("unit", "单位", "如 元/人/天。", searchable=1, editable=1),
|
||
_col("basis_doc", "依据文件", "标准出处文件名，如《差旅费管理办法》。", searchable=1, editable=1),
|
||
_col("clause", "条款", "标准出处条款号，如 第十条。", searchable=1, editable=1),
|
||
_col("effective_date", "生效日期", "标准生效日期（Unix 秒级时间戳）。", searchable=1, editable=1),
|
||
_col("created_at", "创建时间", "记录创建时间（Unix 秒级，自动生成）。"),
|
||
_col("updated_at", "更新时间", "记录最后更新时间（Unix 秒级，自动维护）。"),
|
||
)
|
||
|
||
_TAXID_WHITELIST_COLUMNS: tuple[dict[str, object], ...] = (
|
||
_col("id", "编号", "主键", searchable=1),
|
||
_col("tax_no", "税号", "购买方统一社会信用代码（唯一）。", searchable=1),
|
||
_col("org_name", "单位名称", "税号对应单位全称。", searchable=1, editable=1),
|
||
_col("note", "备注", "维护说明。", searchable=1, editable=1),
|
||
_col("created_at", "创建时间", "记录创建时间（Unix 秒级，自动生成）。"),
|
||
_col("updated_at", "更新时间", "记录最后更新时间（Unix 秒级，自动维护）。"),
|
||
)
|
||
|
||
_HOLIDAY_CALENDAR_COLUMNS: tuple[dict[str, object], ...] = (
|
||
_col("id", "编号", "主键", searchable=1),
|
||
_col("holiday_date", "日期", "节假日日期（YYYY-MM-DD，唯一）。", searchable=1),
|
||
_col("holiday_name", "节假日名称", "如 春节、国庆节。", searchable=1, editable=1),
|
||
_col("created_at", "创建时间", "记录创建时间（Unix 秒级，自动生成）。"),
|
||
_col("updated_at", "更新时间", "记录最后更新时间（Unix 秒级，自动维护）。"),
|
||
)
|
||
|
||
|
||
def _init_invoice_records_metadata(cursor) -> None:
|
||
cursor.execute(_INVOICE_RECORDS_DDL)
|
||
cursor.execute(_INVOICE_RECORDS_INDEX)
|
||
upsert_table_metadata(
|
||
cursor,
|
||
table_name="invoice_records",
|
||
display_name="发票登记",
|
||
description="每次预审自动写入的发票主档，含票面要素与报销上下文。",
|
||
grid_help="按批次号或开票日期查看；本表由预审自动写入，请勿手工修改。",
|
||
visible=1,
|
||
readonly=1,
|
||
default_sort_json='[{"column":"id","direction":"desc"}]',
|
||
default_visible_columns_json='["id","batch_no","invoice_type","invoice_number","issue_date","amount","seller_name","expense_category","project_no"]',
|
||
)
|
||
seed_column_metadata_batch(
|
||
cursor,
|
||
table_name="invoice_records",
|
||
columns=_INVOICE_RECORDS_COLUMNS,
|
||
valid_column_names=[str(c["column_name"]) for c in _INVOICE_RECORDS_COLUMNS],
|
||
)
|
||
|
||
|
||
def _init_prereview_reports_metadata(cursor) -> None:
|
||
cursor.execute(_PREREVIEW_REPORTS_DDL)
|
||
cursor.execute(_PREREVIEW_REPORTS_INDEX)
|
||
upsert_table_metadata(
|
||
cursor,
|
||
table_name="prereview_reports",
|
||
display_name="初审意见",
|
||
description="每张发票的预审结论与红黄绿灯清单；人工查验结果在行级回填。",
|
||
grid_help="“待确认/驳回”的行请在行级按钮回填人工查验结果；HTML 意见单可用顶栏按钮重建。",
|
||
visible=1,
|
||
readonly=1,
|
||
default_sort_json='[{"column":"id","direction":"desc"}]',
|
||
default_visible_columns_json='["id","invoice_number","seller_name","amount","verdict","verify_status","summary","created_at"]',
|
||
)
|
||
seed_column_metadata_batch(
|
||
cursor,
|
||
table_name="prereview_reports",
|
||
columns=_PREREVIEW_REPORTS_COLUMNS,
|
||
valid_column_names=[str(c["column_name"]) for c in _PREREVIEW_REPORTS_COLUMNS],
|
||
)
|
||
|
||
|
||
def _init_expense_standards_metadata(cursor) -> None:
|
||
cursor.execute(_EXPENSE_STANDARDS_DDL)
|
||
init_updated_at_trigger(cursor, "expense_standards")
|
||
upsert_table_metadata(
|
||
cursor,
|
||
table_name="expense_standards",
|
||
display_name="经费标准库",
|
||
description="住宿费上限与交通席位等级标准（城市等级 × 职级），预审硬约束依据。",
|
||
grid_help="数值为示例种子，请以学校现行管理办法为准；可直接编辑标准值/单位/条款。",
|
||
visible=1,
|
||
readonly=0,
|
||
)
|
||
seed_column_metadata_batch(
|
||
cursor,
|
||
table_name="expense_standards",
|
||
columns=_EXPENSE_STANDARDS_COLUMNS,
|
||
valid_column_names=[str(c["column_name"]) for c in _EXPENSE_STANDARDS_COLUMNS],
|
||
)
|
||
|
||
|
||
def _init_taxid_whitelist_metadata(cursor) -> None:
|
||
cursor.execute(_TAXID_WHITELIST_DDL)
|
||
init_updated_at_trigger(cursor, "taxid_whitelist")
|
||
upsert_table_metadata(
|
||
cursor,
|
||
table_name="taxid_whitelist",
|
||
display_name="税号白名单",
|
||
description="准予报销的购买方税号清单；预审时逐张比对，不在名单内即驳回。",
|
||
grid_help="顶栏按钮新增税号；行级按钮可删除错误条目。",
|
||
visible=1,
|
||
readonly=0,
|
||
)
|
||
seed_column_metadata_batch(
|
||
cursor,
|
||
table_name="taxid_whitelist",
|
||
columns=_TAXID_WHITELIST_COLUMNS,
|
||
valid_column_names=[str(c["column_name"]) for c in _TAXID_WHITELIST_COLUMNS],
|
||
)
|
||
|
||
|
||
def _init_holiday_calendar_metadata(cursor) -> None:
|
||
cursor.execute(_HOLIDAY_CALENDAR_DDL)
|
||
init_updated_at_trigger(cursor, "holiday_calendar")
|
||
upsert_table_metadata(
|
||
cursor,
|
||
table_name="holiday_calendar",
|
||
display_name="节假日",
|
||
description="节假日开票风控所用的日历数据（周末之外的法定节假日）。",
|
||
grid_help="以国务院办公厅当年通知为准，请管理员按通知维护连休范围。",
|
||
visible=1,
|
||
readonly=0,
|
||
)
|
||
seed_column_metadata_batch(
|
||
cursor,
|
||
table_name="holiday_calendar",
|
||
columns=_HOLIDAY_CALENDAR_COLUMNS,
|
||
valid_column_names=[str(c["column_name"]) for c in _HOLIDAY_CALENDAR_COLUMNS],
|
||
)
|
||
|
||
|
||
def init_db() -> dict[str, int]:
|
||
"""幂等创建/迁移本地 SQLite、数据管理元数据与业务种子，返回种子统计。"""
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
init_display_metadata(cur)
|
||
_init_invoice_records_metadata(cur)
|
||
_init_prereview_reports_metadata(cur)
|
||
_init_expense_standards_metadata(cur)
|
||
_init_taxid_whitelist_metadata(cur)
|
||
_init_holiday_calendar_metadata(cur)
|
||
seeded = seed_all(cur)
|
||
conn.commit()
|
||
return seeded
|
||
finally:
|
||
conn.close()
|