# /home/Autopartes/pos/services/b2b_service.py """Nexus B2B yonke marketplace: wholesale used parts between deshuesaderos. A yonke (seller) publishes its used/rebuild pieces to a shared catalog; another yonke (buyer) places an order. Listings live in the MASTER DB (nexus_autoparts) so they are visible cross-tenant; the pieces themselves stay in each seller's tenant DB. Tables (master DB): b2b_listings — published pieces (UNIQUE seller_tenant_id+inventory_id) marketplace_orders — order header (buyer/seller tenant ids, total, status) marketplace_order_items — order lines (+ listing_id link) Order lifecycle: pending → confirmed → shipped → received → completed pending/confirmed → cancelled Actor permissions per transition: buyer : cancelled (from pending), received (from confirmed/shipped) seller : confirmed (from pending), shipped (from pending/confirmed), cancelled (from pending/confirmed) both : completed (from received) The source states are deliberately permissive (the UI exposes "Marcar enviado"/"Recibido" a step early) so a valid actor never hits a dead end. Conventions: * Master reads/writes go through the `master_conn` passed by the caller (the blueprint owns commit/rollback), EXCEPT `update_order_status`, which commits the master transaction itself: on `received` it must persist the status BEFORE running the cross-tenant side effects, so the order state can never be rolled back while pieces were already created in another tenant. * Tenant connections are opened/committed/closed inside this module, because one order touches two tenants (buyer + seller) at once. * Everything degrades: side effects at `received` run inside SAVEPOINTs and never abort the order on failure — failures are returned as `warnings`. """ from decimal import Decimal from psycopg2 import errors as pg_errors from psycopg2.extras import RealDictCursor from tenant_db import get_tenant_conn # ─── Domain constants ────────────────────────────────────────────────────── LISTING_STATUSES = ('available', 'reserved', 'sold', 'removed') ORDER_STATUSES = ('pending', 'confirmed', 'shipped', 'received', 'completed', 'cancelled') # status -> {new_status: (allowed actor roles)} ORDER_TRANSITIONS = { 'pending': {'confirmed': ('seller',), 'shipped': ('seller',), 'cancelled': ('buyer', 'seller')}, 'confirmed': {'shipped': ('seller',), 'received': ('buyer',), 'cancelled': ('seller',)}, 'shipped': {'received': ('buyer',)}, 'received': {'completed': ('buyer', 'seller')}, 'completed': {}, 'cancelled': {}, } USED_CONDITIONS = ('usado', 'reconstruido') class B2BError(Exception): """Base class for all B2B errors (mapped to HTTP by the blueprint).""" class B2BValidationError(B2BError): """Bad input / illegal state transition -> HTTP 400.""" class B2BForbidden(B2BError): """Caller is not allowed -> HTTP 403.""" class B2BNotFound(B2BError): """Entity missing -> HTTP 404.""" # ─── Small helpers ───────────────────────────────────────────────────────── def _to_int(val, default=None): if val is None or val == '': return default try: return int(float(str(val).replace(',', ''))) except (TypeError, ValueError): return default def _to_float(val, default=0.0): if val is None or val == '': return default try: return float(str(val).replace(',', '')) except (TypeError, ValueError): return default def _jval(v): """Make a psycopg2 value JSON-safe (Decimal -> float, date -> isoformat).""" if isinstance(v, Decimal): return float(v) if hasattr(v, 'isoformat'): return v.isoformat() return v def _clean(d): return {k: _jval(v) for k, v in (d or {}).items()} def _cur(conn): return conn.cursor(cursor_factory=RealDictCursor) def _tenant_name(master_conn, tenant_id): """Best-effort tenant display name from the master DB (never raises).""" if not tenant_id: return None try: cur = _cur(master_conn) cur.execute("SELECT name FROM tenants WHERE id = %s", (tenant_id,)) row = cur.fetchone() cur.close() return row['name'] if row else None except Exception: try: master_conn.rollback() except Exception: pass return None def is_seller(master_conn, tenant_id): """True when the tenant has the seller flag enabled in the master DB.""" try: cur = _cur(master_conn) cur.execute("SELECT COALESCE(is_seller, false) AS is_seller FROM tenants WHERE id = %s", (tenant_id,)) row = cur.fetchone() cur.close() return bool(row['is_seller']) if row else False except Exception: try: master_conn.rollback() except Exception: pass return False def _audit_best_effort(conn, action, entity_type=None, entity_id=None, new_value=None): try: from services.audit import log_action log_action(conn, action, entity_type, entity_id, new_value=new_value) except Exception: pass # ─── Listings: serialization ─────────────────────────────────────────────── _LISTING_FIELDS = ( 'id', 'seller_tenant_id', 'seller_inventory_id', 'part_number', 'name', 'condition', 'condition_grade', 'price_wholesale', 'donor_make', 'donor_model', 'donor_year', 'image_url', 'status', 'notes', 'created_at', 'updated_at', ) _LISTING_COLS = ', '.join(_LISTING_FIELDS) _LISTING_COLS_L = ', '.join('l.' + f for f in _LISTING_FIELDS) def _listing_dict(row, seller_name=None, include_private=False): """Map a b2b_listings row to the API dict. `inventory_id` is the public alias of `seller_inventory_id` (the contract field). Extra fields (condition, notes, seller_tenant_id, timestamps) are additive and harmless for the UI. """ d = _clean(dict(row)) d['inventory_id'] = d.get('seller_inventory_id') d['price_wholesale'] = float(d.get('price_wholesale') or 0) if seller_name is not None: d['seller_name'] = seller_name if not include_private: d.pop('seller_tenant_id', None) return d # ─── Seller: my listings ─────────────────────────────────────────────────── def list_my_listings(master_conn, tenant_id, status=None): """Listings published by `tenant_id` (removed ones hidden by default).""" cur = _cur(master_conn) where = ["seller_tenant_id = %s"] params = [tenant_id] if status: status = (status or '').strip().lower() if status not in LISTING_STATUSES: raise B2BValidationError(f"status inválido: {status}") where.append("status = %s") params.append(status) else: where.append("status <> 'removed'") cur.execute( f"SELECT {_LISTING_COLS} FROM b2b_listings " f"WHERE {' AND '.join(where)} ORDER BY created_at DESC, id DESC", params, ) rows = cur.fetchall() cur.close() return [_listing_dict(r, include_private=True) for r in rows] def get_listing(master_conn, listing_id): """Single listing + seller_name, or None.""" cur = _cur(master_conn) cur.execute(f"SELECT {_LISTING_COLS} FROM b2b_listings WHERE id = %s", (listing_id,)) row = cur.fetchone() cur.close() if not row: return None return _listing_dict(row, seller_name=_tenant_name(master_conn, row['seller_tenant_id'])) # ─── Seller: publish / edit / remove ─────────────────────────────────────── def _fetch_piece(tenant_id, inventory_id): """Read the source piece from the seller's tenant DB (+ primary image).""" conn = get_tenant_conn(tenant_id) try: cur = conn.cursor(cursor_factory=RealDictCursor) cur.execute(""" SELECT id, part_number, name, condition, condition_grade, donor_make, donor_model, donor_year, image_url, is_active FROM inventory WHERE id = %s """, (inventory_id,)) row = cur.fetchone() if row: cur.execute(""" SELECT url FROM inventory_images WHERE inventory_id = %s ORDER BY is_primary DESC, sort_order ASC, id ASC LIMIT 1 """, (inventory_id,)) img = cur.fetchone() if img and img.get('url'): row = dict(row) row['image_url'] = img['url'] cur.close() return dict(row) if row else None finally: conn.close() def publish_listing(master_conn, tenant_id, data): """Create or refresh a listing for one of the tenant's used pieces. Raises B2BForbidden when the tenant is not a seller, B2BValidationError on bad input / ineligible piece, B2BNotFound when the piece does not exist. """ data = data or {} if not is_seller(master_conn, tenant_id): raise B2BForbidden( 'Tu cuenta no está habilitada como vendedor (is_seller). ' 'Contacta al administrador para activar el modo mayoreo.' ) inventory_id = _to_int(data.get('inventory_id')) if not inventory_id: raise B2BValidationError('inventory_id requerido') if 'price_wholesale' not in data or data.get('price_wholesale') in (None, ''): raise B2BValidationError('price_wholesale requerido') price = _to_float(data.get('price_wholesale'), None) if price is None or price < 0: raise B2BValidationError('price_wholesale inválido') piece = _fetch_piece(tenant_id, inventory_id) if not piece: raise B2BNotFound('Pieza no encontrada') condition = (piece.get('condition') or '').strip().lower() if condition not in USED_CONDITIONS: raise B2BValidationError( "Solo se pueden publicar piezas usadas o reconstruidas " f"(condition actual: {condition or 'desconocida'})" ) if piece.get('is_active') is False: raise B2BValidationError('La pieza no está activa en inventario') notes = (data.get('notes') or '').strip() or None cur = _cur(master_conn) cur.execute(""" INSERT INTO b2b_listings (seller_tenant_id, seller_inventory_id, part_number, name, condition, condition_grade, price_wholesale, donor_make, donor_model, donor_year, image_url, status, notes) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'available', %s) ON CONFLICT (seller_tenant_id, seller_inventory_id) DO UPDATE SET part_number = EXCLUDED.part_number, name = EXCLUDED.name, condition = EXCLUDED.condition, condition_grade = EXCLUDED.condition_grade, price_wholesale = EXCLUDED.price_wholesale, donor_make = EXCLUDED.donor_make, donor_model = EXCLUDED.donor_model, donor_year = EXCLUDED.donor_year, image_url = EXCLUDED.image_url, notes = COALESCE(EXCLUDED.notes, b2b_listings.notes), status = 'available', updated_at = NOW() RETURNING id """, ( tenant_id, inventory_id, piece.get('part_number'), piece.get('name'), condition or 'usado', piece.get('condition_grade'), price, piece.get('donor_make'), piece.get('donor_model'), piece.get('donor_year'), piece.get('image_url'), notes, )) listing_id = cur.fetchone()['id'] cur.close() listing = get_listing(master_conn, listing_id) _audit_publish_best_effort(tenant_id, listing) return listing def _audit_publish_best_effort(tenant_id, listing): try: conn = get_tenant_conn(tenant_id) try: _audit_best_effort(conn, 'B2B_LISTING_PUBLISH', 'b2b_listing', (listing or {}).get('id'), {'price_wholesale': (listing or {}).get('price_wholesale')}) conn.commit() finally: conn.close() except Exception: pass def update_listing(master_conn, tenant_id, listing_id, data): """Update price/status of a listing (owner only). Returns the listing.""" data = data or {} sets = [] params = [] if 'price_wholesale' in data: price = _to_float(data.get('price_wholesale'), None) if price is None or price < 0: raise B2BValidationError('price_wholesale inválido') sets.append('price_wholesale = %s') params.append(price) if 'status' in data: status = (data.get('status') or '').strip().lower() if status not in LISTING_STATUSES: raise B2BValidationError(f"status inválido: {status}") sets.append('status = %s') params.append(status) if 'notes' in data: sets.append('notes = %s') params.append((data.get('notes') or '').strip() or None) if not sets: raise B2BValidationError('Nada que actualizar (price_wholesale, status, notes)') cur = _cur(master_conn) cur.execute("SELECT seller_tenant_id FROM b2b_listings WHERE id = %s", (listing_id,)) row = cur.fetchone() if not row: cur.close() raise B2BNotFound('Listing no encontrado') if row['seller_tenant_id'] != tenant_id: cur.close() raise B2BForbidden('Este listing pertenece a otro vendedor') sets.append('updated_at = NOW()') params.append(listing_id) cur.execute(f"UPDATE b2b_listings SET {', '.join(sets)} WHERE id = %s", params) cur.close() return get_listing(master_conn, listing_id) def remove_listing(master_conn, tenant_id, listing_id): """Soft-delete a listing (status='removed'). Owner only. Returns bool.""" cur = _cur(master_conn) cur.execute("SELECT seller_tenant_id FROM b2b_listings WHERE id = %s", (listing_id,)) row = cur.fetchone() if not row: cur.close() raise B2BNotFound('Listing no encontrado') if row['seller_tenant_id'] != tenant_id: cur.close() raise B2BForbidden('Este listing pertenece a otro vendedor') cur.execute( "UPDATE b2b_listings SET status = 'removed', updated_at = NOW() WHERE id = %s", (listing_id,), ) cur.close() return True # ─── Buyer: catalog ──────────────────────────────────────────────────────── def list_catalog(master_conn, buyer_tenant_id, filters=None, page=1, per_page=12): """Available listings from OTHER tenants, with seller_name and pagination.""" filters = filters or {} page = max(_to_int(page, 1) or 1, 1) per_page = min(max(_to_int(per_page, 12) or 12, 1), 100) where = ["l.status = 'available'", 'l.seller_tenant_id <> %s'] params = [buyer_tenant_id] q = (filters.get('q') or '').strip() if q: where.append("(l.name ILIKE %s OR l.part_number ILIKE %s " "OR l.donor_make ILIKE %s OR l.donor_model ILIKE %s)") params.extend([f'%{q}%'] * 4) for field, col in (('make', 'l.donor_make'), ('model', 'l.donor_model')): val = (filters.get(field) or '').strip() if val: where.append(f"{col} ILIKE %s") params.append(f'%{val}%') year = _to_int(filters.get('year')) if year: where.append("l.donor_year = %s") params.append(year) grade = (filters.get('grade') or '').strip().upper() if grade: where.append("UPPER(l.condition_grade) = %s") params.append(grade) where_sql = ' AND '.join(where) cur = _cur(master_conn) cur.execute(f""" SELECT COUNT(*) AS total FROM b2b_listings l WHERE {where_sql} """, params) total = cur.fetchone()['total'] or 0 cur.execute(f""" SELECT {_LISTING_COLS_L}, t.name AS seller_name FROM b2b_listings l LEFT JOIN tenants t ON t.id = l.seller_tenant_id WHERE {where_sql} ORDER BY l.created_at DESC, l.id DESC LIMIT %s OFFSET %s """, params + [per_page, (page - 1) * per_page]) rows = cur.fetchall() cur.close() items = [_listing_dict(r, seller_name=r.get('seller_name')) for r in rows] pagination = { 'page': page, 'per_page': per_page, 'total': total, 'total_pages': (total + per_page - 1) // per_page, } return items, pagination # ─── Orders ──────────────────────────────────────────────────────────────── _ORDER_COLS = ( 'id, buyer_tenant_id, seller_tenant_id, buyer_name, seller_name, total, ' 'status, notes, created_at, updated_at' ) def _order_items(master_conn, order_ids): """Bulk-load order lines grouped by order_id.""" if not order_ids: return {} cur = _cur(master_conn) cur.execute(""" SELECT id, order_id, listing_id, part_number, part_name, quantity, unit_price, subtotal FROM marketplace_order_items WHERE order_id = ANY(%s) ORDER BY order_id, id """, (list(order_ids),)) rows = cur.fetchall() cur.close() out = {} for r in rows: item = _clean(dict(r)) item['name'] = item.get('part_name') # UI-friendly alias out.setdefault(r['order_id'], []).append(item) return out def _order_dict(row, items, role): d = _clean(dict(row)) d['total'] = float(d.get('total') or 0) d['role'] = role if role == 'buyer': d['other_party'] = d.get('seller_name') d['other_tenant_id'] = d.get('seller_tenant_id') else: d['other_party'] = d.get('buyer_name') d['other_tenant_id'] = d.get('buyer_tenant_id') d['items'] = items or [] return d def create_order(master_conn, buyer_tenant_id, items, notes=None): """Create a pending order + items and reserve the listings. All lines must reference available listings from ONE other tenant. """ if not isinstance(items, list) or not items: raise B2BValidationError('items requerido (lista no vacía)') wanted = [] for it in items: if not isinstance(it, dict): raise B2BValidationError('cada item debe ser un objeto {listing_id, quantity}') listing_id = _to_int(it.get('listing_id')) qty = _to_int(it.get('quantity'), 1) if not listing_id: raise B2BValidationError('listing_id requerido en cada item') if qty is None or qty < 1: raise B2BValidationError('quantity debe ser >= 1') wanted.append({'listing_id': listing_id, 'quantity': qty}) listing_ids = [w['listing_id'] for w in wanted] cur = _cur(master_conn) cur.execute( f"SELECT {_LISTING_COLS} FROM b2b_listings WHERE id = ANY(%s) FOR UPDATE", (listing_ids,), ) rows = {r['id']: dict(r) for r in cur.fetchall()} cur.close() sellers = set() resolved = [] for w in wanted: listing = rows.get(w['listing_id']) if not listing: raise B2BNotFound(f"Listing {w['listing_id']} no encontrado") if listing['status'] != 'available': raise B2BValidationError( f"Listing {w['listing_id']} no está disponible (status: {listing['status']})") if listing['seller_tenant_id'] == buyer_tenant_id: raise B2BValidationError('No puedes comprar tus propias piezas') sellers.add(listing['seller_tenant_id']) resolved.append((listing, w['quantity'])) if len(sellers) > 1: raise B2BValidationError( 'Una orden solo puede tener un vendedor; crea una orden por vendedor') seller_tenant_id = sellers.pop() total = round(sum( _to_float(listing['price_wholesale'], 0) * qty for listing, qty in resolved), 2) buyer_name = _tenant_name(master_conn, buyer_tenant_id) or '' seller_name = _tenant_name(master_conn, seller_tenant_id) or '' cur = _cur(master_conn) cur.execute(""" INSERT INTO marketplace_orders (buyer_tenant_id, seller_tenant_id, buyer_name, seller_name, total, status, notes) VALUES (%s, %s, %s, %s, %s, 'pending', %s) RETURNING id """, (buyer_tenant_id, seller_tenant_id, buyer_name, seller_name, total, (notes or '').strip() or None)) order_id = cur.fetchone()['id'] order_items = [] for listing, qty in resolved: price = _to_float(listing['price_wholesale'], 0) subtotal = round(price * qty, 2) cur.execute(""" INSERT INTO marketplace_order_items (order_id, listing_id, part_number, part_name, quantity, unit_price, subtotal) VALUES (%s, %s, %s, %s, %s, %s, %s) """, (order_id, listing['id'], listing.get('part_number'), listing.get('name'), qty, price, subtotal)) order_items.append({ 'listing_id': listing['id'], 'part_number': listing.get('part_number'), 'part_name': listing.get('name'), 'name': listing.get('name'), 'quantity': qty, 'unit_price': price, 'subtotal': subtotal, }) cur.execute(""" UPDATE b2b_listings SET status = 'reserved', updated_at = NOW() WHERE id = ANY(%s) """, ([listing['id'] for listing, _ in resolved],)) cur.close() return { 'id': order_id, 'total': total, 'status': 'pending', 'seller_tenant_id': seller_tenant_id, 'buyer_tenant_id': buyer_tenant_id, 'seller_name': seller_name, 'buyer_name': buyer_name, 'items': order_items, } def list_orders(master_conn, tenant_id, role): """Orders where `tenant_id` is buyer ('buyer') or seller ('seller').""" col = 'buyer_tenant_id' if role == 'buyer' else 'seller_tenant_id' cur = _cur(master_conn) cur.execute( f"SELECT {_ORDER_COLS} FROM marketplace_orders WHERE {col} = %s " f"ORDER BY created_at DESC, id DESC", (tenant_id,), ) rows = cur.fetchall() cur.close() items = _order_items(master_conn, [r['id'] for r in rows]) return [_order_dict(r, items.get(r['id']), role) for r in rows] def get_order(master_conn, tenant_id, order_id): """Order detail; only participants. Raises B2BNotFound / B2BForbidden.""" cur = _cur(master_conn) cur.execute(f"SELECT {_ORDER_COLS} FROM marketplace_orders WHERE id = %s", (order_id,)) row = cur.fetchone() cur.close() if not row: raise B2BNotFound('Orden no encontrada') if row['buyer_tenant_id'] == tenant_id: role = 'buyer' elif row['seller_tenant_id'] == tenant_id: role = 'seller' else: raise B2BForbidden('No participas en esta orden') items = _order_items(master_conn, [order_id]) return _order_dict(row, items.get(order_id), role) def update_order_status(master_conn, tenant_id, order_id, new_status, branch_id=None): """Apply an order status transition (role-checked). Commits the master transaction itself: when moving to `received` the status must be durable before the cross-tenant side effects (marking the seller's piece sold and creating the piece in the buyer's inventory) run. """ new_status = (new_status or '').strip().lower() if new_status not in ORDER_STATUSES: raise B2BValidationError( f"status inválido: {new_status} (válidos: {', '.join(ORDER_STATUSES)})") cur = _cur(master_conn) cur.execute( f"SELECT {_ORDER_COLS} FROM marketplace_orders WHERE id = %s FOR UPDATE", (order_id,), ) row = cur.fetchone() if not row: cur.close() raise B2BNotFound('Orden no encontrada') if row['buyer_tenant_id'] == tenant_id: role = 'buyer' elif row['seller_tenant_id'] == tenant_id: role = 'seller' else: cur.close() raise B2BForbidden('No participas en esta orden') current = row['status'] or 'pending' allowed = ORDER_TRANSITIONS.get(current, {}).get(new_status) if not allowed: cur.close() raise B2BValidationError(f"No se puede pasar de '{current}' a '{new_status}'") if role not in allowed: cur.close() raise B2BForbidden( f"Rol '{role}' no puede cambiar la orden de '{current}' a '{new_status}'") cur.execute( "UPDATE marketplace_orders SET status = %s, updated_at = NOW() WHERE id = %s", (new_status, order_id), ) cur.close() master_conn.commit() result = get_order(master_conn, tenant_id, order_id) result['previous_status'] = current if new_status == 'received': extras = _finalize_received(master_conn, dict(row), branch_id=branch_id) result.update(extras) return result # ─── Received: cross-tenant side effects (best-effort) ───────────────────── def _resolve_branch(conn, branch_id=None): if branch_id: return branch_id try: cur = conn.cursor() cur.execute("SELECT id FROM branches ORDER BY id LIMIT 1") r = cur.fetchone() cur.close() return r[0] if r else None except Exception: return None def _mark_seller_piece_sold(conn, inventory_id): """Mark the seller's inventory row as sold (SAVEPOINT, never raises). `inventory` has no `status` column in the standard tenant schema, so we set `is_active = false` (piece no longer available) when it is absent; if a tenant ever adds `inventory.status`, we write 'sold' there instead. """ cur = conn.cursor() cur.execute("SAVEPOINT b2b_sold") try: cur.execute(""" SELECT 1 FROM information_schema.columns WHERE table_name = 'inventory' AND column_name = 'status' LIMIT 1 """) has_status = cur.fetchone() is not None if has_status: cur.execute("UPDATE inventory SET status = 'sold' WHERE id = %s", (inventory_id,)) else: cur.execute("UPDATE inventory SET is_active = false WHERE id = %s", (inventory_id,)) cur.execute("RELEASE SAVEPOINT b2b_sold") return True except Exception: try: cur.execute("ROLLBACK TO SAVEPOINT b2b_sold") except Exception: pass return False finally: cur.close() def _insert_buyer_inventory(conn, item, listing, unit_price, warnings): """Insert the purchased piece into the buyer's inventory (SAVEPOINT).""" from services.yonke_service import generate_unique_sku name = (listing.get('name') or item.get('part_name') or item.get('part_number') or 'Pieza usada') candidate = (listing.get('part_number') or item.get('part_number') or '').strip() or None grade = listing.get('condition_grade') cur = conn.cursor() inv_id = None for _attempt in range(5): if candidate is None: try: candidate = generate_unique_sku(conn, None) except Exception: candidate = None cur.execute("SAVEPOINT b2b_inv") try: cur.execute(""" INSERT INTO inventory (part_number, name, unit, cost, price_1, price_2, price_3, tax_rate, condition, condition_grade, is_unique_piece, donor_make, donor_model, donor_year, ownership, image_url, is_active) VALUES (%s, %s, 'PZA', %s, %s, %s, %s, %s, 'usado', %s, true, %s, %s, %s, 'propio', %s, true) RETURNING id """, ( candidate, name, unit_price, unit_price, unit_price, unit_price, 0.16, grade, listing.get('donor_make'), listing.get('donor_model'), listing.get('donor_year'), listing.get('image_url'), )) inv_id = cur.fetchone()[0] cur.execute("RELEASE SAVEPOINT b2b_inv") break except pg_errors.UniqueViolation: cur.execute("ROLLBACK TO SAVEPOINT b2b_inv") candidate = None except Exception as e: try: cur.execute("ROLLBACK TO SAVEPOINT b2b_inv") except Exception: pass warnings.append(f"no se pudo crear la pieza en el inventario del comprador: {e}") cur.close() return None cur.close() return inv_id def _seed_stock(conn, inv_id, branch_id, qty, cost, warnings): """Seed initial stock for a freshly created buyer piece (SAVEPOINT).""" if not branch_id or qty <= 0: return from services.inventory_engine import record_initial cur = conn.cursor() cur.execute("SAVEPOINT b2b_stock") try: record_initial(conn, inv_id, branch_id, qty, cost=cost) cur.execute("RELEASE SAVEPOINT b2b_stock") except Exception as e: try: cur.execute("ROLLBACK TO SAVEPOINT b2b_stock") except Exception: pass warnings.append(f"no se registró la operación de stock: {e}") # Idempotent safety net: the trigger normally creates this row already. try: cur.execute(""" INSERT INTO inventory_stock (inventory_id, branch_id, stock) VALUES (%s, %s, %s) ON CONFLICT (inventory_id, branch_id) DO UPDATE SET stock = EXCLUDED.stock """, (inv_id, branch_id, qty)) except Exception as e: warnings.append(f"no se pudo fijar el stock inicial: {e}") cur.close() def _finalize_received(master_conn, order, branch_id=None): """On `received`: sell the listings/seller pieces and stock the buyer. Best-effort by design: every step runs in its own SAVEPOINT / connection, failures are collected in `warnings`, and the order is never aborted. """ warnings = [] extras = {'listings_sold': 0, 'inventory_created': 0, 'warnings': warnings} order_id = order['id'] seller_id = order['seller_tenant_id'] buyer_id = order['buyer_tenant_id'] cur = _cur(master_conn) cur.execute(""" SELECT id, listing_id, part_number, part_name, quantity, unit_price FROM marketplace_order_items WHERE order_id = %s ORDER BY id """, (order_id,)) items = [dict(r) for r in cur.fetchall()] listing_ids = [it['listing_id'] for it in items if it['listing_id']] listings = {} if listing_ids: cur.execute(f"SELECT {_LISTING_COLS} FROM b2b_listings WHERE id = ANY(%s)", (listing_ids,)) for r in cur.fetchall(): listings[r['id']] = dict(r) # Legacy rows may lack listing_id: fall back to (seller, part_number). for it in items: if it['listing_id'] in listings: continue cur.execute(f""" SELECT {_LISTING_COLS} FROM b2b_listings WHERE seller_tenant_id = %s AND part_number = %s ORDER BY (status = 'reserved') DESC, id DESC LIMIT 1 """, (seller_id, it.get('part_number'))) r = cur.fetchone() if r: listings[it['listing_id']] = dict(r) # 1) Mark listings sold (master DB) ------------------------------------ sold = 0 for it in items: listing = listings.get(it['listing_id']) if it['listing_id'] else None if not listing and it['listing_id']: continue try: if listing: cur.execute( "UPDATE b2b_listings SET status = 'sold', updated_at = NOW() WHERE id = %s", (listing['id'],)) else: cur.execute(""" UPDATE b2b_listings SET status = 'sold', updated_at = NOW() WHERE seller_tenant_id = %s AND part_number = %s AND status = 'reserved' """, (seller_id, it.get('part_number'))) sold += cur.rowcount except Exception as e: master_conn.rollback() warnings.append(f"no se pudo marcar el listing como vendido: {e}") cur = _cur(master_conn) try: master_conn.commit() except Exception as e: warnings.append(f"no se pudo confirmar el estado de los listings: {e}") extras['listings_sold'] = sold cur.close() # 2) Mark the seller's pieces sold (seller tenant DB) ------------------ seller_conn = None try: seller_conn = get_tenant_conn(seller_id) for it in items: listing = listings.get(it['listing_id']) inv_id = _to_int((listing or {}).get('seller_inventory_id')) if inv_id: _mark_seller_piece_sold(seller_conn, inv_id) seller_conn.commit() except Exception as e: warnings.append(f"no se pudieron marcar las piezas del vendedor: {e}") if seller_conn is not None: try: seller_conn.rollback() except Exception: pass finally: if seller_conn is not None: try: seller_conn.close() except Exception: pass # 3) Create the purchased pieces in the buyer's inventory -------------- buyer_conn = None try: buyer_conn = get_tenant_conn(buyer_id) buyer_branch = _resolve_branch(buyer_conn, branch_id) created = 0 for it in items: listing = listings.get(it['listing_id']) or {} qty = max(_to_int(it.get('quantity'), 1) or 1, 1) price = _to_float(it.get('unit_price'), 0) inv_id = _insert_buyer_inventory(buyer_conn, it, listing, price, warnings) if inv_id: _seed_stock(buyer_conn, inv_id, buyer_branch, qty, price, warnings) created += 1 buyer_conn.commit() extras['inventory_created'] = created try: _audit_best_effort(buyer_conn, 'B2B_ORDER_RECEIVED', 'marketplace_order', order_id, {'inventory_created': created}) buyer_conn.commit() except Exception: pass except Exception as e: warnings.append(f"no se pudo crear el inventario del comprador: {e}") if buyer_conn is not None: try: buyer_conn.rollback() except Exception: pass finally: if buyer_conn is not None: try: buyer_conn.close() except Exception: pass return extras # ─── Seller status (UI badge) ────────────────────────────────────────────── def get_seller_status(master_conn, tenant_id): """{is_seller, listings_count, incoming_orders} for the current tenant.""" seller = is_seller(master_conn, tenant_id) cur = _cur(master_conn) cur.execute(""" SELECT COUNT(*) AS c FROM b2b_listings WHERE seller_tenant_id = %s AND status <> 'removed' """, (tenant_id,)) listings_count = cur.fetchone()['c'] or 0 cur.execute(""" SELECT COUNT(*) AS c FROM marketplace_orders WHERE seller_tenant_id = %s AND status IN ('pending', 'confirmed', 'shipped') """, (tenant_id,)) incoming = cur.fetchone()['c'] or 0 cur.execute(""" SELECT COUNT(*) AS c FROM marketplace_orders WHERE seller_tenant_id = %s """, (tenant_id,)) total_incoming = cur.fetchone()['c'] or 0 cur.close() return { 'is_seller': seller, 'listings_count': listings_count, 'incoming_orders': incoming, 'incoming_orders_total': total_incoming, }