fix(address+ui): remove TIER 2 reuse, typed diff badges, false positive reduction

- Remove TIER 2 address lookup (county+city without street) from PL/SQL — creates
  new address when street differs instead of reusing wrong one
- Replace generic "N diferente" badge with typed micro-badges (CUI, Denumire, TVA,
  Adr. livr., Adr. fact., Preturi) with red/amber semantic colors
- Extend addrMatch() regex to strip full Romanian address words (STRADA, NUMAR, BLOC,
  COMUNA, SAT, MUNICIPIUL, etc.) — fixes "Strada X" vs "X" false positives
- Extend normalize_company_name() for II, PFA, INTREPRINDERE INDIVIDUALA legal forms
- Persist address_mismatch to SQLite so "Dif." filter includes address-only diffs
- Add red/amber indicator dots to desktop table and mobile list rows
- 12 unit tests for normalization and server-side address matching

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Agent
2026-04-06 14:28:57 +00:00
parent fc1013bff6
commit 31095c07f7
11 changed files with 160 additions and 44 deletions

View File

@@ -1,6 +1,8 @@
import asyncio
import json
import logging
import re
import unicodedata
import uuid
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
@@ -18,6 +20,34 @@ from .. import database
logger = logging.getLogger(__name__)
def _addr_match(gomag_json, roa_json):
"""Server-side address comparison matching JS addrMatch()."""
if not gomag_json or not roa_json:
return True
try:
g = json.loads(gomag_json) if isinstance(gomag_json, str) else gomag_json
r = json.loads(roa_json) if isinstance(roa_json, str) else roa_json
except (json.JSONDecodeError, TypeError):
return True
_ADDR_WORDS = re.compile(
r'\b(STR|STRADA|NR|NUMAR|NUMARUL|BL|BLOC|SC|SCARA|AP|APART|APARTAMENT|'
r'ET|ETAJ|COM|COMUNA|SAT|MUN|MUNICIPIUL|JUD|JUDETUL|CARTIER|PARTER|SECTOR|ORAS)\b'
)
def norm(s):
s = unicodedata.normalize('NFD', s or '')
s = re.sub(r'[\u0300-\u036f]', '', s).upper()
s = _ADDR_WORDS.sub('', s)
return re.sub(r'[^A-Z0-9]', '', s)
g_street = norm(g.get('address') or g.get('strada') or '')
r_street = norm((r.get('strada') or '') + (r.get('numar') or ''))
g_city = norm(g.get('city') or g.get('localitate') or '')
r_city = norm(r.get('localitate') or '')
g_region = norm(g.get('region') or g.get('judet') or '')
r_region = norm(r.get('judet') or '')
return g_street == r_street and g_city == r_city and g_region == r_region
# Sync state
_sync_lock = asyncio.Lock()
_current_sync = None # dict with run_id, status, progress info
@@ -802,6 +832,11 @@ async def run_sync(id_pol: int = None, id_sectie: int = None, run_id: str = None
partner_data["anaf_denumire_mismatch"] = 1
partner_data["denumire_anaf"] = anaf_data_for_order["denumire_anaf"]
# Address mismatch check (server-side, mirrors JS addrMatch)
livr_match = _addr_match(partner_data.get("adresa_livrare_gomag"), partner_data.get("adresa_livrare_roa"))
fact_match = _addr_match(partner_data.get("adresa_facturare_gomag"), partner_data.get("adresa_facturare_roa"))
partner_data["address_mismatch"] = 1 if (not livr_match or not fact_match) else 0
await sqlite_service.update_order_partner_data(order.number, partner_data)
if not result["success"]: