35 lines
1.2 KiB
Python
35 lines
1.2 KiB
Python
"""Promote ORIGEM_MERCADORIA/CST_ICMS/CSOSN/CST_PIS/CST_COFINS from the local
|
|
official MOC (Anexo I NF-e/NFC-e) PDF. No network fetch — this manual has no
|
|
stable public direct-download URL either (session-gated portal), so the
|
|
authoritative copy lives in sources/raw/bootstrap_v1/.
|
|
"""
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from central.app.datasets.nfe_moc_tables import parse_moc_tables
|
|
from central.app.datasets.service import DatasetService
|
|
from central.app.db import SessionLocal
|
|
|
|
SOURCE = Path("sources/raw/bootstrap_v1/moc7-anexo-i-leiaute-e-rv.pdf")
|
|
SOURCE_URL = f"file://{SOURCE.name}"
|
|
|
|
|
|
def main():
|
|
raw_bytes = SOURCE.read_bytes()
|
|
payload = parse_moc_tables(raw_bytes)
|
|
|
|
with SessionLocal() as session:
|
|
dataset = DatasetService(session).import_moc_tables(payload, SOURCE_URL, datetime.now(timezone.utc))
|
|
|
|
from collections import Counter
|
|
domains = Counter(e.domain for e in payload.entries)
|
|
print(json.dumps({
|
|
"status": dataset.status, "dataset_id": dataset.id, "sha256": dataset.sha256,
|
|
"record_count": dataset.record_count, "domains": dict(domains),
|
|
}, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|