Un flux zilnic de firme noi, fără să pierzi niciuna

Jobul de dimineață a rulat fără nicio eroare toată luna, și totuși cineva din vânzări a dat singur peste o firmă nouă din Cluj care nu ajunsese niciodată în CRM. Jobul cerea „firmele înregistrate ieri”; firma fusese înregistrată joi și indexată luni. Nu crăpase nimic, doar întrebarea era greșită.

Mai jos e un job în Python care ia în fiecare zi firmele noi de la /registrations, le pune într-o bază SQLite și poate fi oprit, repornit sau uitat o săptămână fără să piardă ori să dubleze vreuna. Codul e scurt; contează cele trei decizii din spatele lui.

Trei date care nu sunt același lucru

Fiecare firmă din listă are două momente. registered_on este ziua înregistrării: intrarea în registrul comerțului sau, dacă aceea nu se știe încă, înregistrarea fiscală la ANAF. discovered_at este momentul în care firma a intrat în baza apifirme. ANAF atribuie codurile fiscale în ordine, apifirme parcurge șirul din oră în oră, iar o firmă ajunge de obicei în baza de date în ziua următoare înregistrării. „De obicei” înseamnă că uneori durează câteva zile, peste un weekend sau când ANAF răspunde greu.

Al treilea moment este ceasul serverului tău, și e cel mai tentant: „dă-mi tot ce s-a indexat în ultimele 24 de ore”. Merge până în ziua în care cronul pornește cu câteva minute mai târziu decât ieri, în care cineva trimite ora României cu Z la coadă (trei ore diferență vara) sau în care jobul nu rulează deloc. Fiecare lasă o fâșie de timp pe care n-o mai cere nimeni, iar firmele din fâșia aceea nu dau nicio eroare. Pur și simplu lipsesc.

Regula care nu lasă găuri, așa cum o descrie și documentația: după ultima pagină, ia discovered_at al ultimei firme primite și trimite-l la rularea următoare ca discovered_after. Firmele devin vizibile strict în ordinea discovered_at, deci tot ce vine după acel moment e nou pentru tine. Dacă jobul a stat trei zile, a patra zi ia trei zile deodată.

O cerere de mână

Înainte de program, aceeași întrebare cu curl și jq. Răspunsurile de aici și de mai jos vin de pe un server de test, cu date de exemplu:

$ curl -s -G https://apifirme.dev/rest/v1/registrations \
    -H "Authorization: Bearer $KEY" \
    -d discovered_after=2026-10-02T12:00:00Z -d limit=3 |
  jq -r '(.data[] | "\(.cui) \(.registered_on) \(.discovered_at) \(.caen)"), .next_cursor'
91003264 2026-09-29 2026-10-02T12:46:52Z 1071
91003132 2026-09-25 2026-10-02T13:33:42Z 4941
91003680 2026-10-01 2026-10-02T16:30:55Z 4639
bzoz

Am cerut firmele indexate după 2 octombrie, ora 12:00 UTC, iar a doua coloană arată înregistrări din 25 septembrie până pe 1 octombrie. Un job care ar fi cerut registered_after=2026-10-01 ar fi văzut doar ultima firmă. Ultimul rând e next_cursor: cât timp nu e null, mai sunt pagini, și îl trimiți ca cursor împreună cu aceiași parametri.

Un detaliu mic: păstrează discovered_at ca text, exact cum a venit. Trecut printr-un parser de date și scris la loc, își poate schimba precizia sau fusul orar: o fracțiune de secundă mai devreme aduce înapoi o firmă deja primită, una mai târziu poate sări una.

Aceeași firmă, a doua oară

A doua firmă din listă, 91003132, nu e la prima apariție:

$ curl -s -G https://apifirme.dev/rest/v1/registrations \
    -H "Authorization: Bearer $KEY" -d limit=500 |
  jq -r '.data[] | select(.cui == 91003132) | "\(.discovered_at) caen=\(.caen)"'
2026-09-29T13:33:42Z caen=null
2026-10-02T13:33:42Z caen=4941

Prima dată a venit fără cod CAEN. Nu e o greșeală: un sfert dintre firmele noi nu au încă un cod CAEN la ANAF în prima zi (despre ce se știe și ce nu în primele zile scrie în articolul despre fereastra de decizie a firmelor noi). Când codul apare, firma reintră în listă cu un discovered_at nou. La fel se întâmplă când i se află județul sau forma juridică.

Două consecințe pentru cod. Prima: firmele se scriu cu upsert după cui, niciodată cu un simplu insert, altfel ai dubluri sau o eroare de cheie la a doua apariție. A doua: dacă filtrezi după caen sau caen_prefix, firma ajunge la tine abia la a doua apariție, pentru că la prima nu corespundea filtrului. E exact ce vrei, dar „nouă pentru mine” nu mai înseamnă „înregistrată ieri”; de aceea programul ține separat vazuta_prima_data.

Programul

Python 3.8 sau mai nou, requests și sqlite3 din biblioteca standard. Cheia vine din APIFIRME_KEY; FIRME_NOI_JUDET (de exemplu SB,BV) e un filtru opțional.

#!/usr/bin/env python3
"""Ia firmele noi de la apifirme într-o bază SQLite, fără găuri și fără dubluri."""
import json
import os
import sqlite3
import sys
import time
from datetime import datetime, timezone

import requests

BASE = os.environ.get("APIFIRME_BASE", "https://apifirme.dev/rest/v1")
KEY = os.environ["APIFIRME_KEY"]
DB = os.environ.get("FIRME_NOI_DB", "firme_noi.db")
PAGINA = int(os.environ.get("FIRME_NOI_PAGINA", "500"))  # între 1 și 500
# Filtre opționale, ca în documentație: judet, caen, caen_prefix, forma_juridica.
FILTRE = {"judet": os.environ.get("FIRME_NOI_JUDET")}

SCHEMA = """
create table if not exists firme (
  cui integer primary key,
  denumire text, judet text, caen integer, forma_juridica_cod text,
  registered_on text, discovered_at text not null,
  vazuta_prima_data text not null, vazuta_ultima_data text not null,
  brut text not null
);
create table if not exists stare (cheie text primary key, valoare text not null);
"""

UPSERT = """
insert into firme values (:cui, :denumire, :judet, :caen, :forma_juridica_cod,
  :registered_on, :discovered_at, :acum, :acum, :brut)
on conflict (cui) do update set
  denumire = excluded.denumire, judet = excluded.judet, caen = excluded.caen,
  forma_juridica_cod = excluded.forma_juridica_cod,
  registered_on = excluded.registered_on, discovered_at = excluded.discovered_at,
  vazuta_ultima_data = excluded.vazuta_ultima_data, brut = excluded.brut
"""
CAMPURI = ("cui", "denumire", "judet", "caen", "forma_juridica_cod",
           "registered_on", "discovered_at")


def cere_pagina(sesiune, params):
    for incercare in range(5):
        try:
            r = sesiune.get(f"{BASE}/registrations", params=params, timeout=(5, 30))
        except requests.RequestException as e:
            print(f"  încercarea {incercare + 1}: {type(e).__name__}", file=sys.stderr)
            time.sleep(2 ** incercare)
            continue
        if r.status_code == 200:
            return r.json()
        if r.status_code == 429 or r.status_code >= 500:
            pauza = int(r.headers.get("Retry-After", 2 ** incercare))
            print(f"  încercarea {incercare + 1}: {r.status_code}, aștept {pauza} s",
                  file=sys.stderr)
            time.sleep(min(pauza, 60))
            continue
        # 400, 401, 403 (planul nu include lista): repetarea nu ajută
        p = r.json() if "json" in r.headers.get("Content-Type", "") else {}
        sys.exit(f"{r.status_code} {p.get('title', '')}: {p.get('detail', r.text[:200])}")
    sys.exit("apifirme nu răspunde acum; starea a rămas neschimbată, reia mai târziu")


def salveaza(db, firme):
    """Upsert după CUI. Întoarce (noi, actualizate)."""
    noi = 0
    acum = datetime.now(timezone.utc).isoformat(timespec="seconds")
    for f in firme:
        if not db.execute("select 1 from firme where cui = ?", (f["cui"],)).fetchone():
            noi += 1
        rand = {k: f.get(k) for k in CAMPURI}
        rand.update(acum=acum, brut=json.dumps(f, ensure_ascii=False))
        db.execute(UPSERT, rand)
    return noi, len(firme) - noi


def main():
    db = sqlite3.connect(DB)
    db.executescript(SCHEMA)
    rand = db.execute("select valoare from stare where cheie = 'discovered_after'")
    start = (rand.fetchone() or [None])[0]

    params = {k: v for k, v in FILTRE.items() if v}
    params["limit"] = PAGINA
    if start:
        params["discovered_after"] = start  # textul primit, nu reformatat

    total_noi = total_actualizate = nr = 0
    ultimul = None
    with requests.Session() as sesiune:
        sesiune.headers["Authorization"] = f"Bearer {KEY}"
        while True:
            pagina = cere_pagina(sesiune, params)
            firme, nr = pagina["data"], nr + 1
            with db:  # o tranzacție pe pagină
                noi, actualizate = salveaza(db, firme)
                if firme:
                    ultimul = firme[-1]["discovered_at"]
                if pagina["next_cursor"] is None and ultimul:
                    # abia după ultima pagină, în aceeași tranzacție cu ultimele firme
                    db.execute("insert or replace into stare values "
                               "('discovered_after', ?)", (ultimul,))
            total_noi += noi
            total_actualizate += actualizate
            print(f"pagina {nr}: {len(firme)} firme")
            if pagina["next_cursor"] is None:
                break
            params["cursor"] = pagina["next_cursor"]

    print(f"{total_noi} firme noi, {total_actualizate} actualizate, "
          f"discovered_after = {ultimul or start or '(nimic încă)'}")


if __name__ == "__main__":
    main()

Programul nu trimite days, așa că perioada cerută este toată fereastra planului tău. La prima rulare, fără stare salvată, asta înseamnă toate firmele din fereastră, adică câteva sute pe fiecare zi lucrătoare din ea, în pagini de câte 500. Coloana brut păstrează răspunsul întreg, pentru câmpurile pe care nu le-ai scos în tabel.

De ce arată așa

Starea se scrie abia după ultima pagină

Ar părea mai prudent să salvezi discovered_after după fiecare pagină. Problema apare când două firme au exact același discovered_at și granița paginii cade între ele: discovered_after înseamnă strict „după”, deci a doua s-ar pierde. next_cursor nu are problema asta, pentru că reia exact de unde s-a oprit pagina. Așa că în timpul unei rulări mergi cu cursorul, iar punctul de plecare pentru data viitoare îl scrii o singură dată, la sfârșit.

Dacă jobul cade la pagina a patra din zece, starea a rămas cea de ieri. Rularea următoare reia de acolo, primește din nou primele trei pagini, iar upsert-ul le înghite fără dubluri. Am verificat asta pe serverul de test: am oprit forțat programul la pagina a patra, l-am pornit din nou, iar baza a ieșit identică, firmă cu firmă, cu lista completă luată dintr-o singură cerere.

O tranzacție pe pagină

with db: deschide o tranzacție și o confirmă la ieșire. Ultima pagină și noua stare intră în aceeași tranzacție, deci nu poți ajunge cu firmele salvate și starea veche sau invers. Dacă ții starea într-un fișier separat, scrie-o într-un fișier temporar și mută-l peste cel vechi cu os.replace; un fișier pe jumătate scris, citit a doua zi, e cel mai urât fel de a pierde firme.

Ce se repetă și ce nu

Se repetă doar ce poate merge la a doua încercare: 429 și erorile 5xx, după cât spune Retry-After, plus conexiunile căzute, cu pauze care cresc. Un 400, un 401 sau un 403 opresc programul cu mesajul din corpul erorii (formatul e descris la erori). 403 înseamnă că planul cheii nu include lista firmelor noi; numărul lor, /registrations/count, e inclus în orice plan, iar planurile cu listă sunt pe pagina de prețuri.

Rulat pe serverul de test

Prima rulare, cu pagini mici ca să se vadă paginarea, apoi a doua, imediat după:

$ FIRME_NOI_PAGINA=50 python3 flux_firme_noi.py
pagina 1: 50 firme
pagina 2: 50 firme
pagina 3: 50 firme
pagina 4: 37 firme
154 firme noi, 33 actualizate, discovered_after = 2026-10-03T18:18:15Z
$ python3 flux_firme_noi.py
pagina 1: 0 firme
0 firme noi, 0 actualizate, discovered_after = 2026-10-03T18:18:15Z
$ sqlite3 firme_noi.db "select count(*), count(caen) from firme"
154|150

187 de înregistrări, 154 de firme: 33 au venit a doua oară în aceeași fereastră, după ce li s-a aflat codul CAEN. Patru sunt încă fără cod și vor reveni. A doua rulare primește o pagină goală și lasă starea neschimbată. Apoi, cu serverul de test pus să închidă conexiunile:

$ python3 flux_firme_noi.py; echo "cod de ieșire $?"
  încercarea 1: ConnectionError
  încercarea 2: ConnectionError
  încercarea 3: ConnectionError
  încercarea 4: ConnectionError
  încercarea 5: ConnectionError
apifirme nu răspunde acum; starea a rămas neschimbată, reia mai târziu
cod de ieșire 1

Codul de ieșire diferit de zero e pentru cron și pentru monitorizarea ta; starea a rămas cea de dinainte, deci data viitoare jobul ia tot ce a pierdut azi.

În cron

# crontab -e: în fiecare zi la 7:30, o singură instanță odată
APIFIRME_KEY=af_live_...
30 7 * * * cd /opt/firme-noi && flock -n .lock python3 flux_firme_noi.py >> flux.log 2>&1

flock -n nu lasă două rulări să se suprapună când una se lungește. O dată pe zi e suficient pentru un CRM; dacă vrei firmele mai repede, câteva rulări pe zi au sens, dar mai des decât din oră în oră nu aduci nimic, pentru că lista se completează din oră în oră. Ține minte că și o pagină goală e un răspuns 200 și se numără ca cerere.

Mai pune o alarmă simplă: dacă discovered_after din tabelul stare nu s-a mișcat de două zile lucrătoare, ceva e stricat. Fără filtre, zilnic apar câteva sute de firme noi, deci o stare înghețată nu e liniște, e un job care nu mai merge. Cu filtre înguste (un județ mic și un singur cod CAEN) pragul trebuie mai larg. Iar un job oprit mai mult decât fereastra planului pierde de-a binelea firmele înregistrate înainte de ea, pentru că nu mai încap în nicio cerere.

De încercat

Rulează programul de două ori la rând pe cheia ta. A doua rulare trebuie să spună 0 firme noi. Apoi șterge rândul din stare și rulează-l din nou cu FIRME_NOI_PAGINA=20: numărul de firme din bază trebuie să rămână exact același. Dacă se schimbă, ai găsit o gaură sau o dublură înainte s-o găsească cineva din vânzări.