182 lines
6.8 KiB
Python
182 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
監査・移行用エクスポートツール
|
||
- SQLite DB から伝票単位で JSON Lines を出力
|
||
- ハッシュチェーン検証結果と説明文も出力
|
||
"""
|
||
|
||
import json
|
||
import sqlite3
|
||
import sys
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from typing import List, Dict, Any
|
||
|
||
def export_invoices_as_jsonl(db_path: str, out_path: str) -> None:
|
||
"""DBから伝票を1行JSONでエクスポート"""
|
||
conn = sqlite3.connect(db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
cur = conn.cursor()
|
||
|
||
cur.execute("""
|
||
SELECT
|
||
i.uuid,
|
||
i.document_type,
|
||
i.customer_name,
|
||
i.customer_address,
|
||
i.customer_phone,
|
||
i.amount,
|
||
i.tax,
|
||
i.total_amount,
|
||
i.date,
|
||
i.invoice_number,
|
||
i.notes,
|
||
i.node_id,
|
||
i.payload_json,
|
||
i.payload_hash,
|
||
i.prev_chain_hash,
|
||
i.chain_hash,
|
||
i.pdf_template_version,
|
||
i.company_info_version,
|
||
i.is_offset,
|
||
i.offset_target_uuid,
|
||
i.pdf_generated_at,
|
||
i.pdf_sha256,
|
||
GROUP_CONCAT(
|
||
json_object(
|
||
'description', ii.description,
|
||
'quantity', ii.quantity,
|
||
'unit_price', ii.unit_price,
|
||
'is_discount', ii.is_discount
|
||
), ','
|
||
) AS items_json
|
||
FROM invoices i
|
||
LEFT JOIN invoice_items ii ON i.id = ii.invoice_id
|
||
GROUP BY i.id
|
||
ORDER BY i.id ASC
|
||
""")
|
||
|
||
with open(out_path, "w", encoding="utf-8") as f:
|
||
for row in cur:
|
||
items = json.loads(row["items_json"] or "[]")
|
||
record = {
|
||
"uuid": row["uuid"],
|
||
"document_type": row["document_type"],
|
||
"customer": {
|
||
"name": row["customer_name"],
|
||
"address": row["customer_address"],
|
||
"phone": row["customer_phone"],
|
||
},
|
||
"amount": row["amount"],
|
||
"tax": row["tax"],
|
||
"total_amount": row["total_amount"],
|
||
"date": row["date"],
|
||
"invoice_number": row["invoice_number"],
|
||
"notes": row["notes"],
|
||
"node_id": row["node_id"],
|
||
"payload_json": json.loads(row["payload_json"]) if row["payload_json"] else None,
|
||
"payload_hash": row["payload_hash"],
|
||
"prev_chain_hash": row["prev_chain_hash"],
|
||
"chain_hash": row["chain_hash"],
|
||
"pdf_template_version": row["pdf_template_version"],
|
||
"company_info_version": row["company_info_version"],
|
||
"is_offset": bool(row["is_offset"]),
|
||
"offset_target_uuid": row["offset_target_uuid"],
|
||
"pdf_generated_at": row["pdf_generated_at"],
|
||
"pdf_sha256": row["pdf_sha256"],
|
||
"items": items,
|
||
}
|
||
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||
conn.close()
|
||
|
||
def generate_audit_summary(db_path: str) -> Dict[str, Any]:
|
||
"""ハッシュチェーン検証とサマリーを返す"""
|
||
from services.repositories import InvoiceRepository
|
||
repo = InvoiceRepository(db_path)
|
||
result = repo.verify_chain()
|
||
# 件数
|
||
conn = sqlite3.connect(db_path)
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT COUNT(*) FROM invoices")
|
||
total = cur.fetchone()[0]
|
||
cur.execute("SELECT COUNT(*) FROM invoices WHERE is_offset=1")
|
||
offsets = cur.fetchone()[0]
|
||
conn.close()
|
||
return {
|
||
"generated_at": datetime.now().isoformat(),
|
||
"db_path": db_path,
|
||
"total_invoices": total,
|
||
"offset_invoices": offsets,
|
||
"chain_verification": result,
|
||
}
|
||
|
||
def write_audit_readme(out_dir: Path) -> None:
|
||
"""監査用説明文（README）を出力"""
|
||
readme = """# 監査・移出パッケージ
|
||
|
||
## 概要
|
||
本パッケージは「販売アシスト1号」のSQLite DBからエクスポートした伝票データと、ハッシュチェーン検証結果を含みます。
|
||
|
||
## ファイル
|
||
- `invoices.jsonl` : 伝票1件ごとに1行のJSON（payload_json含む）
|
||
- `audit_summary.json` : 全件数・赤伝件数・チェーン検証結果
|
||
- `README.md` : 本ファイル
|
||
|
||
## ハッシュチェーン方式（改ざん検知）
|
||
- 各伝票は `payload_json` を正規化JSONとし、その SHA256 を `payload_hash` として保持
|
||
- `chain_hash = SHA256(prev_chain_hash + ':' + payload_hash)` で直前伝票と連鎖
|
||
- 先頭伝票の `prev_chain_hash` は 64文字のゼロ（genesis）
|
||
- `node_id` はDB単位のUUIDで、端末故障時の復旧・識別に使用
|
||
|
||
## 検証手順
|
||
1. `audit_summary.json` の `chain_verification.ok` が `true` であること
|
||
2. `false` の場合は `chain_verification.errors` を確認
|
||
3. 任意：スクリプトで `invoices.jsonl` を読み込み、先頭から `chain_hash` を再計算して一致を確認
|
||
|
||
## 再現性（PDF生成）
|
||
- `payload_json` と `pdf_template_version`/`company_info_version` があれば、
|
||
同一内容のPDFを再生成可能（PDFバイナリの完全一致は保証しない）
|
||
- PDFは永続保存せず、必要時に再生成して利用・共有・削除する方針
|
||
|
||
## 赤伝（相殺）の扱い
|
||
- `is_offset=true` の伝票は赤伝（相殺伝票）
|
||
- `offset_target_uuid` で元伝票を指す
|
||
- 集計・表示ではデフォルト除外（UIの「赤伝を表示」スイッチで切替）
|
||
|
||
## 10年保管について
|
||
- SQLite DB（本エクスポート元）を正とし、10年間保管
|
||
- バックアップ・復旧はDBファイル単位で行う（Google Drive等）
|
||
- 必要に応じて本エクスポートパッケージを外部システムへ移行可能
|
||
|
||
---
|
||
生成日時: {generated_at}
|
||
""".format(generated_at=datetime.now().isoformat())
|
||
(out_dir / "README.md").write_text(readme, encoding="utf-8")
|
||
|
||
def main():
|
||
db_path = "sales.db"
|
||
out_dir = Path("audit_export")
|
||
out_dir.mkdir(exist_ok=True)
|
||
jsonl_path = out_dir / "invoices.jsonl"
|
||
summary_path = out_dir / "audit_summary.json"
|
||
|
||
print("Exporting invoices as JSONL...")
|
||
export_invoices_as_jsonl(db_path, str(jsonl_path))
|
||
print(f" -> {jsonl_path}")
|
||
|
||
print("Generating audit summary...")
|
||
summary = generate_audit_summary(db_path)
|
||
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f" -> {summary_path}")
|
||
|
||
print("Writing README...")
|
||
write_audit_readme(out_dir)
|
||
print(f" -> {out_dir}/README.md")
|
||
|
||
print("\nAudit export complete.")
|
||
print(f"Total invoices: {summary['total_invoices']}")
|
||
print(f"Offset invoices: {summary['offset_invoices']}")
|
||
print(f"Chain verification OK: {summary['chain_verification']['ok']}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|