134 lines
9.1 KiB
Python
134 lines
9.1 KiB
Python
"""Persistent guard for workflow drafts; no database lock spans external awaits."""
|
|
import copy
|
|
from decimal import Decimal
|
|
import aiohttp
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from psycopg2.extras import Json
|
|
from app.core.config import settings
|
|
from app.core.auth_dependencies import require_permission
|
|
from .service import transaction, json_safe
|
|
|
|
router=APIRouter(prefix='/workflows/exports')
|
|
|
|
def is_workflow_draft(draft_id):
|
|
if not getattr(settings,'WORKFLOWS_ENABLED',False) or not draft_id: return False
|
|
with transaction() as cursor:
|
|
cursor.execute('SELECT 1 FROM bmc_workflow_settlements WHERE draft_id=%s LIMIT 1',(draft_id,))
|
|
return bool(cursor.fetchone())
|
|
|
|
def claim(draft_id,request):
|
|
with transaction() as cursor:
|
|
cursor.execute('INSERT INTO bmc_workflow_exports(draft_id,request,created_by) VALUES(%s,%s,%s) ON CONFLICT(draft_id) DO NOTHING',(draft_id,Json(json_safe(request)),request.get('user_id')))
|
|
cursor.execute('SELECT * FROM bmc_workflow_exports WHERE draft_id=%s FOR UPDATE',(draft_id,))
|
|
row=dict(cursor.fetchone())
|
|
if row['status']=='done': return row
|
|
if row['status'] in ('sending','uncertain'):
|
|
raise HTTPException(409,'Eksporten behandles eller er uafklaret. Afstem den eksisterende kladde; force_export tillader ikke et nyt forsøg.')
|
|
cursor.execute('SELECT sync_status,economic_order_number,customer_id FROM ordre_drafts WHERE id=%s',(draft_id,))
|
|
draft=cursor.fetchone()
|
|
if draft and str(draft.get('customer_id'))!=str(request.get('customer_id')):
|
|
raise HTTPException(409,'Eksportkunden skal være den samme som workflowkladdens kunde')
|
|
if draft and (draft.get('economic_order_number') or draft.get('sync_status') in ('exported','posted','paid')):
|
|
raise HTTPException(409,'Kladden er allerede eksporteret. Afstem den eksisterende eksport før fortsættelse.')
|
|
cursor.execute("UPDATE bmc_workflow_exports SET status='sending',request=%s,error=NULL,updated_at=now() WHERE draft_id=%s",(Json(json_safe(request)),draft_id))
|
|
row['status']='sending'
|
|
return row
|
|
|
|
def record(draft_id,status,result=None,error=None):
|
|
with transaction() as cursor:
|
|
cursor.execute('UPDATE bmc_workflow_exports SET status=%s,result=%s,error=%s,updated_at=now() WHERE draft_id=%s',(status,Json(json_safe(result)) if result is not None else None,error,draft_id))
|
|
|
|
def recovered_result(draft_id):
|
|
with transaction() as cursor:
|
|
cursor.execute("SELECT result FROM bmc_workflow_exports WHERE draft_id=%s AND status='done'",(draft_id,))
|
|
row=cursor.fetchone()
|
|
return copy.deepcopy(row['result']) if row else None
|
|
|
|
async def guarded_export(draft_id,exporter,request):
|
|
if not is_workflow_draft(draft_id): return await exporter(**request)
|
|
attempt=claim(draft_id,request)
|
|
if attempt['status']=='done': return copy.deepcopy(attempt['result'])
|
|
outbound=dict(request,export_reference=attempt['reference'])
|
|
try:
|
|
result=await exporter(**outbound)
|
|
result['workflow_export_reference']=attempt['reference']
|
|
result['exported_line_keys']=[line['line_key'] for line in request.get('lines',[]) if line.get('selected',True) and line.get('line_key')]
|
|
if result.get('dry_run'):
|
|
record(draft_id,'pending',result)
|
|
else:
|
|
number=result.get('economic_order_number') or result.get('economic_draft_id') or result.get('orderNumber')
|
|
if not result.get('success',True) or not number: raise HTTPException(502,'Eksportsvaret mangler et verificerbart ordrenummer')
|
|
# Commit the remote response before the caller updates its local draft.
|
|
record(draft_id,'done',result)
|
|
return result
|
|
except BaseException as exc:
|
|
definitive = isinstance(exc,HTTPException) and exc.status_code in (400,403,404,409,422)
|
|
record(draft_id,'pending' if definitive else 'uncertain',error=str(exc)[:1000])
|
|
raise
|
|
|
|
def verify_remote(attempt,remote,expected_customer,expected_products):
|
|
request=attempt['request']
|
|
if str((remote.get('references') or {}).get('other'))!=str(attempt['draft_id']): raise HTTPException(409,'Ordren har ikke kladdens præcise Hub-reference')
|
|
if str((remote.get('customer') or {}).get('customerNumber'))!=str(expected_customer): raise HTTPException(409,'Kunden afviger')
|
|
if remote.get('currency')!=request.get('currency','DKK'): raise HTTPException(409,'Valutaen afviger')
|
|
expected=[line for line in request['lines'] if line.get('selected',True)]
|
|
actual=remote.get('lines') or []
|
|
if len(actual)!=len(expected): raise HTTPException(409,'Antallet af ordrelinjer afviger')
|
|
for line,match,product in zip(expected,actual,expected_products):
|
|
if str((match.get('product') or {}).get('productNumber') or '')!=str(product or ''): raise HTTPException(409,'Varenummeret afviger')
|
|
if (match.get('description') or '')!=(line.get('description') or 'Ordrelinje'): raise HTTPException(409,'Beskrivelsen afviger')
|
|
for local,external in (('quantity','quantity'),('unit_price','unitNetPrice'),('discount_percentage','discountPercentage')):
|
|
if Decimal(str(line.get(local) or 0))!=Decimal(str(match.get(external) or 0)): raise HTTPException(409,'Ordrelinjens værdier afviger')
|
|
|
|
class ReconcileRequest(BaseModel):
|
|
economic_order_number:int=Field(gt=0)
|
|
|
|
@router.get('')
|
|
def attempts(user=Depends(require_permission('economic.documents.export'))):
|
|
with transaction() as cursor:
|
|
cursor.execute('SELECT draft_id,reference,status,error,created_at,updated_at FROM bmc_workflow_exports ORDER BY updated_at DESC LIMIT 200')
|
|
return cursor.fetchall()
|
|
|
|
@router.post('/{draft_id}/reconcile')
|
|
async def reconcile(draft_id:int,payload:ReconcileRequest,user=Depends(require_permission('economic.documents.export')),admin=Depends(require_permission('workflows.manage'))):
|
|
with transaction() as cursor:
|
|
cursor.execute('SELECT * FROM bmc_workflow_exports WHERE draft_id=%s',(draft_id,))
|
|
row=cursor.fetchone()
|
|
if not row: raise HTTPException(404,'Eksportforsøg findes ikke')
|
|
attempt=dict(row)
|
|
if attempt['status']=='done': return attempt['result']
|
|
from app.products.backend.economic_documents import active_connection
|
|
connection=active_connection()
|
|
if connection:
|
|
from app.products.backend.economic_catalog import query
|
|
from app.products.backend.economic_documents import reconcile as reconcile_catalog
|
|
existing=query('SELECT id FROM economic_document_exports WHERE connection_id=%s AND kind=%s AND document_key=%s',(connection['id'],'order',f'order-draft:{draft_id}'),one=True)
|
|
if not existing: raise HTTPException(409,'Fandt ingen eksisterende ekstern eksportidentitet; ingen ny eksport tilladt')
|
|
result=await reconcile_catalog(existing['id'])
|
|
if str(result.get('economic_order_number') or result.get('economic_draft_id'))!=str(payload.economic_order_number): raise HTTPException(409,'Det afstemte ordrenummer afviger')
|
|
else:
|
|
from app.modules.orders.backend.economic_export import ordre_economic_export_service as service
|
|
request=attempt['request']
|
|
with transaction() as cursor:
|
|
cursor.execute('SELECT economic_customer_number FROM customers WHERE id=%s',(request['customer_id'],))
|
|
customer=cursor.fetchone()
|
|
if not customer or not customer['economic_customer_number']: raise HTTPException(409,'Kundens e-conomic-identitet mangler')
|
|
products=[]
|
|
for line in request['lines']:
|
|
if not line.get('selected',True): continue
|
|
cursor.execute('SELECT sku_internal,economic_product_number FROM products WHERE id=%s',(line.get('product_id'),))
|
|
product=cursor.fetchone() or {}
|
|
number=(product.get('economic_product_number') if line.get('source_type')=='invoice_fee' else product.get('sku_internal')) or service.default_product
|
|
products.append(number)
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(f'{service.api_url}/orders/drafts/{payload.economic_order_number}',headers=service._headers(),timeout=aiohttp.ClientTimeout(total=30)) as response:
|
|
if response.status!=200: raise HTTPException(409,'Den eksisterende ordre kunne ikke verificeres i e-conomic')
|
|
remote=await response.json()
|
|
verify_remote(attempt,remote,customer['economic_customer_number'],products)
|
|
result={'success':True,'dry_run':False,'economic_order_number':payload.economic_order_number,'economic_draft_id':payload.economic_order_number,'details':{'economic_response':remote,'request_payload':remote,'economic_customer_number':customer['economic_customer_number'],'currency':remote.get('currency'),'economic_order_url':remote.get('self')},'message':'Eksisterende kladde afstemt; ingen ny ordre oprettet'}
|
|
result['workflow_export_reference']=attempt['reference']
|
|
result['exported_line_keys']=[line['line_key'] for line in attempt['request'].get('lines',[]) if line.get('selected',True) and line.get('line_key')]
|
|
record(draft_id,'done',result)
|
|
return result
|