83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Migra as notas da skill (tabela `notas`) para `fiscal_documents` do app.
|
|
|
|
Idempotente: deduplica por (mes, ano, supplier_name, total_paid).
|
|
|
|
Uso:
|
|
python scripts/migrate_notas.py [caminho_do_banco_origem]
|
|
|
|
Origem padrão: Skill/dados/lernotafiscal.db
|
|
Destino: o banco do app (DB_PATH / data/app.sqlite3).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# permite rodar de qualquer lugar
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from app import database as db # noqa: E402
|
|
from app.dates import resolve_competencia # noqa: E402
|
|
|
|
DEFAULT_SOURCE = Path(__file__).resolve().parent.parent / "Skill" / "dados" / "lernotafiscal.db"
|
|
|
|
|
|
def read_notas(source: Path) -> list[tuple]:
|
|
conn = sqlite3.connect(source)
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT data_compra, fornecedor, valor_pago, arquivo_origem FROM notas"
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
return [(r["data_compra"], r["fornecedor"], float(r["valor_pago"]), r["arquivo_origem"]) for r in rows]
|
|
|
|
|
|
def main() -> int:
|
|
source = Path(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_SOURCE
|
|
if not source.exists():
|
|
print(f"Banco de origem não encontrado: {source}")
|
|
return 1
|
|
|
|
notas = read_notas(source)
|
|
inserted = skipped = 0
|
|
with db.session() as conn:
|
|
for data_compra, fornecedor, valor, arquivo in notas:
|
|
mes, ano = resolve_competencia(data_compra)
|
|
exists = conn.execute(
|
|
"""
|
|
SELECT 1 FROM fiscal_documents
|
|
WHERE mes = ? AND ano = ? AND supplier_name = ? AND ROUND(total_paid, 2) = ROUND(?, 2)
|
|
LIMIT 1
|
|
""",
|
|
(mes, ano, fornecedor, valor),
|
|
).fetchone()
|
|
if exists:
|
|
skipped += 1
|
|
continue
|
|
db.create_fiscal(
|
|
conn,
|
|
mes=mes,
|
|
ano=ano,
|
|
supplier_name=fornecedor,
|
|
total_paid=valor,
|
|
source_file_name=arquivo or "migrado",
|
|
source_location="migrado",
|
|
confidence="high",
|
|
)
|
|
inserted += 1
|
|
totals = db.overall_totals(conn)
|
|
|
|
print(f"Origem: {source}")
|
|
print(f"Inseridos: {inserted} · Ignorados (duplicados): {skipped}")
|
|
print(f"Total em fiscal_documents: {totals['count']} documento(s) · R$ {totals['total']:.2f}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|