diff --git a/app/customers/backend/merge.py b/app/customers/backend/merge.py new file mode 100644 index 0000000..35f21a4 --- /dev/null +++ b/app/customers/backend/merge.py @@ -0,0 +1,309 @@ +"""Local, atomic customer consolidation; never writes to external systems. + +Relations are discovered from the actual schema, including legacy integer Hub +references without foreign keys. Conflicts abort the entire operation. Preview +executes the same transaction and rolls it back, including trigger effects. +""" +import hashlib +import json +import logging +from contextlib import contextmanager + +import psycopg2 +from psycopg2 import sql +from psycopg2.extras import RealDictCursor, Json +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, Field + +from app.core.database import get_db_connection, release_db_connection +from app.core.auth_dependencies import require_permission +from app.core.config import settings + +logger = logging.getLogger(__name__) +router = APIRouter() + + +class MergeSelection(BaseModel): + target_id: int = Field(gt=0) + source_ids: list[int] = Field(min_length=1, max_length=50) + + +class MergeConfirmation(MergeSelection): + preview_token: str = Field(min_length=64, max_length=64) + + +@contextmanager +def transaction(commit=False): + conn = get_db_connection() + try: + with conn.cursor(cursor_factory=RealDictCursor) as cursor: + cursor.execute("SET LOCAL lock_timeout = '5s'") + cursor.execute("SET LOCAL statement_timeout = '30s'") + yield cursor + if commit: + conn.commit() + else: + conn.rollback() + except BaseException: + conn.rollback() + raise + finally: + release_db_connection(conn) + + +def selections(target, sources): + sources = sorted(set(sources)) + if not sources or target in sources or any(i <= 0 for i in sources): + raise HTTPException(422, 'Vælg en hovedkunde og mindst én anden kunde.') + return sources + + +def relations(cur): + """Only real tables in current schema; never confuse tmodule customer IDs. + + A numeric customer reference with an FK to another entity is excluded. + Partition children are handled through their parent, avoiding double moves. + """ + cur.execute(""" + SELECT c.relname AS table_name, a.attname AS column_name, + EXISTS(SELECT 1 FROM pg_attribute x WHERE x.attrelid=c.oid + AND x.attname='entity_type' AND NOT x.attisdropped) AS polymorphic, + EXISTS(SELECT 1 FROM pg_constraint f WHERE f.conrelid=c.oid AND f.contype='f' + AND a.attnum=ANY(f.conkey) AND f.confrelid='customers'::regclass) AS customer_fk, + EXISTS(SELECT 1 FROM pg_constraint f WHERE f.conrelid=c.oid AND f.contype='f' + AND a.attnum=ANY(f.conkey) AND f.confrelid<>'customers'::regclass) AS other_fk + FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace + JOIN pg_attribute a ON a.attrelid=c.oid AND a.attnum>0 AND NOT a.attisdropped + WHERE n.nspname=current_schema() AND c.relkind IN ('r','p') AND NOT c.relispartition + AND c.relname<>'customer_merge_history' + AND a.atttypid IN ('int2'::regtype,'int4'::regtype,'int8'::regtype) + AND (a.attname='customer_id' OR a.attname LIKE '%%_customer_id' + OR a.attname='entity_id' OR EXISTS( + SELECT 1 FROM pg_constraint f WHERE f.conrelid=c.oid AND f.contype='f' + AND a.attnum=ANY(f.conkey) AND f.confrelid='customers'::regclass)) + ORDER BY c.relname,a.attname + """) + return [r for r in cur.fetchall() if not r['other_fk'] + and r['column_name'] != 'merged_into_customer_id' + and (r['column_name'] != 'entity_id' or r['polymorphic']) + # These logs use IDs from the time module, not Hub IDs. + and not (r['table_name'] in ('tmodule_sync_log', 'tticket_audit_log') + and r['column_name'] == 'entity_id')] + + +def where(ref): + clause = sql.SQL('{} = ANY(%s)').format(sql.Identifier(ref['column_name'])) + if ref['column_name'] == 'entity_id': + clause += sql.SQL(" AND entity_type='customer'") + return clause + + +def signature(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest() + + +def mentions_customer(value, sources): + if isinstance(value,list): + return any(mentions_customer(v,sources) for v in value) + if not isinstance(value,dict): + return False + candidates=[value.get('customer_id')] + if value.get('field') == 'customer_id': + candidates.extend(value.get('value') if isinstance(value.get('value'),list) else [value.get('value')]) + return any(str(v) in {str(i) for i in sources} for v in candidates if v is not None) or any( + mentions_customer(v,sources) for v in value.values() if isinstance(v,(dict,list))) + + +IDENTITY_FIELDS = ('economic_customer_number', 'vtiger_id', 'vtiger_account_no', 'cvr_number') +IDENTITY_LABELS = {'economic_customer_number':'e-conomic-kundenumre','vtiger_id':'vTiger-id’er', + 'vtiger_account_no':'vTiger-kundenumre','cvr_number':'CVR-numre'} +FILL_FIELDS = ('email', 'phone', 'address', 'email_domain', 'city', 'postal_code', + 'country', 'website', 'invoice_email', 'mobile_phone', 'wiki_slug') + + +def field_plan(customers, target): + changes, differences = {}, [] + for field in IDENTITY_FIELDS + FILL_FIELDS: + values = {str(c[field]).strip(): c[field] for c in customers if c.get(field) not in (None, '')} + if field in IDENTITY_FIELDS and len(values) > 1: + raise HTTPException(409, f'Kunderne har forskellige {IDENTITY_LABELS[field]}. Afklar relationen før sammenlægning.') + if not target.get(field) and len(values) == 1: + changes[field] = next(iter(values.values())) + elif len(values) > 1: + differences.append(field) + excluded=set(IDENTITY_FIELDS+FILL_FIELDS) | {'id','created_at','updated_at','deleted_at','last_synced_at', + 'merged_into_customer_id','parent_customer_id','billing_customer_id'} + for field in sorted(set().union(*(c.keys() for c in customers))-excluded): + values={json.dumps(c.get(field),default=str,sort_keys=True) for c in customers} + if len(values)>1: differences.append(field) + return changes, differences + + +def prepare(cur, target_id, source_ids): + cur.execute("SELECT to_regclass('customer_merge_history') AS history") + if not cur.fetchone()['history']: + raise HTTPException(503, 'Kundesammenlægning kræver migration 1074_customer_merge.sql.') + # Consistent locking order for merges; FK inserts are serialized with these locks. + cur.execute('SELECT * FROM customers WHERE id=ANY(%s) ORDER BY id FOR UPDATE', + (sorted([target_id] + source_ids),)) + customers = [dict(r) for r in cur.fetchall()] + if len(customers) != len(source_ids) + 1: + raise HTTPException(404, 'En valgt kunde findes ikke længere.') + if any(c.get('merged_into_customer_id') or c.get('deleted_at') or c.get('is_active') is False for c in customers): + raise HTTPException(409, 'En valgt kunde er allerede samlet, slettet eller deaktiveret. Genindlæs listen.') + target = next(c for c in customers if c['id'] == target_id) + changes, differences = field_plan(customers, target) + refs = relations(cur) + # Lock relation tables so legacy references without FKs cannot change mid-merge. + tables = {r['table_name'] for r in refs} | {'customers'} + for table in ('bmc_workflows','bmc_workflow_versions','bmc_workflow_events','bmc_workflow_runtime'): + cur.execute('SELECT to_regclass(%s) AS name',(table,)) + if cur.fetchone()['name']: tables.add(table) + for table in sorted(tables): + cur.execute(sql.SQL('LOCK TABLE {} IN SHARE ROW EXCLUSIVE MODE').format(sql.Identifier(table))) + state, movements, origins = [], [], [] + for ref in refs: + cur.execute(sql.SQL('SELECT * FROM {} WHERE {}').format(sql.Identifier(ref['table_name']),where(ref)), + ([target_id]+source_ids,)) + rows = [dict(r) for r in cur.fetchall()] + if ref['table_name']=='ordre_drafts': + economic=changes.get('economic_customer_number',target.get('economic_customer_number')) + if any(r.get('exported_economic_customer_number') not in (None,'') and + str(r['exported_economic_customer_number']).strip()!=str(economic).strip() for r in rows): + raise HTTPException(409,'En ordre eller fakturakladde er eksporteret til et andet e-conomic-kundenummer. Afklar fakturarelationen før sammenlægning.') + if rows: + state.append({'table':ref['table_name'],'column':ref['column_name'], + 'rows':sorted(rows,key=lambda r:json.dumps(r,sort_keys=True,default=str))}) + cur.execute('''SELECT a.attname FROM pg_index i JOIN pg_attribute a + ON a.attrelid=i.indrelid AND a.attnum=ANY(i.indkey) + WHERE i.indrelid=%s::regclass AND i.indisprimary ORDER BY a.attname''', + (ref['table_name'],)) + keys=[r['attname'] for r in cur.fetchall()] + if not keys: + raise HTTPException(409,f"Tabellen {ref['table_name']} mangler en entydig nøgle til flyttehistorikken.") + origins.extend({'table':ref['table_name'],'column':ref['column_name'], + 'key':{k:r[k] for k in keys},'original_customer_id':r[ref['column_name']]} + for r in rows if r[ref['column_name']] in source_ids) + if ref['table_name'] in ('contact_companies','entity_tags'): + origins.append({'table':ref['table_name'],'original_memberships':rows}) + count = sum(r[ref['column_name']] in source_ids for r in rows) + if count: + movements.append(dict(ref, count=count)) + # bmc_workflows stores Hub IDs as text, so numeric FK discovery cannot see it. + cur.execute("SELECT to_regclass('bmc_workflows') AS name") + if cur.fetchone()['name']: + cur.execute('SELECT * FROM bmc_workflows WHERE customer_id=ANY(%s) ORDER BY id', + ([str(i) for i in [target_id]+source_ids],)) + rows=[dict(r) for r in cur.fetchall()] + state.append({'table':'bmc_workflows','rows':rows}) + origins.extend({'table':'bmc_workflows','column':'customer_id','key':{'id':r['id']}, + 'original_customer_id':r['customer_id']} for r in rows + if r['customer_id'] in [str(i) for i in source_ids]) + count=sum(r['customer_id'] in [str(i) for i in source_ids] for r in rows) + if count: movements.append(dict(table_name='bmc_workflows',column_name='customer_id',count=count)) + cur.execute("SELECT to_regclass('bmc_workflow_runtime') AS name") + if cur.fetchone()['name']: + cur.execute('SELECT customer_ids FROM bmc_workflow_runtime') + if any(set(r['customer_ids'] or []) & set(source_ids) for r in cur.fetchall()): + raise HTTPException(409, 'En kildekunde er valgt i workflowmotorens kundeliste. Afklar aktiveringen før sammenlægning.') + if set(getattr(settings,'WORKFLOWS_CUSTOMER_IDS',[]) or []) & set(source_ids): + raise HTTPException(409,'En kildekunde er valgt i workflowkonfigurationen. Afklar aktiveringen før sammenlægning.') + for table in ('bmc_workflow_events','bmc_workflow_versions'): + cur.execute('SELECT to_regclass(%s) AS name',(table,)) + if not cur.fetchone()['name']: continue + if table == 'bmc_workflow_events': + cur.execute("SELECT context,entity_type,entity_id FROM bmc_workflow_events WHERE status IN ('pending','processing','failed')") + blocked=any(mentions_customer(r['context'],source_ids) or + (r['entity_type']=='customer' and str(r['entity_id']) in {str(i) for i in source_ids}) for r in cur.fetchall()) + else: + cur.execute('''SELECT v.definition FROM bmc_workflow_versions v JOIN bmc_workflows w + ON w.active_version_id=v.id WHERE w.enabled''') + blocked=any(mentions_customer(r['definition'],source_ids) for r in cur.fetchall()) + if blocked: + raise HTTPException(409,'En kildekunde bruges i aktive workflowregler eller uafsluttede job. Afklar dem før sammenlægning.') + token=signature({'customers':customers,'state':state,'target':target_id,'sources':source_ids}) + return customers, changes, differences, movements, token, origins + + +def move(cur, target, sources, customers, changes, movements): + """Any unhandled unique constraint is a blocking conflict, never data loss.""" + archived=[] + # Transfer unique external IDs only after saving original customer snapshots. + for field in IDENTITY_FIELDS: + if field in customers[0] and (field != 'cvr_number' or field in changes): + cur.execute(sql.SQL('UPDATE customers SET {}=NULL WHERE id=ANY(%s)').format(sql.Identifier(field)),(sources,)) + for field,value in changes.items(): + cur.execute(sql.SQL('UPDATE customers SET {}=%s WHERE id=%s').format(sql.Identifier(field)),(value,target)) + if any(r['table_name']=='customer_relationships' for r in movements): + cur.execute('''DELETE FROM customer_relationships WHERE parent_customer_id=ANY(%s) + AND child_customer_id=ANY(%s) RETURNING *''',([target]+sources,[target]+sources)) + archived.extend({'table':'customer_relationships','row':dict(r)} for r in cur.fetchall()) + # Shared tag/contact memberships are consolidated; retain their complete old rows in audit. + for table,key,condition in [('contact_companies','contact_id',''),('entity_tags','tag_id'," AND entity_type='customer'")]: + if not any(r['table_name']==table for r in movements): continue + column='entity_id' if table=='entity_tags' else 'customer_id' + cur.execute(sql.SQL('SELECT * FROM {} WHERE {}=ANY(%s)'+condition+' ORDER BY {}').format( + sql.Identifier(table),sql.Identifier(column),sql.Identifier('id')),([target]+sources,)) + rows=[dict(r) for r in cur.fetchall()];seen={} + for row in sorted(rows,key=lambda r:(r[column]!=target,r['id'])): + if row[key] not in seen: + seen[row[key]]=row;continue + kept=seen[row[key]] + archived.append({'table':table,'row':row}) + if table=='contact_companies': + notes='\n'.join(dict.fromkeys(str(v) for v in (kept.get('notes'),row.get('notes')) if v)) + roles=' / '.join(dict.fromkeys(str(v) for v in (kept.get('role'),row.get('role')) if v)) + cur.execute('UPDATE contact_companies SET notes=%s,role=%s,is_primary=%s WHERE id=%s', + (notes or None,roles or None,bool(kept.get('is_primary') or row.get('is_primary')),kept['id'])) + kept.update(notes=notes,role=roles,is_primary=bool(kept.get('is_primary') or row.get('is_primary'))) + cur.execute(sql.SQL('DELETE FROM {} WHERE id=%s').format(sql.Identifier(table)),(row['id'],)) + if table=='contact_companies' and any(r[column]==target and r.get('is_primary') for r in rows): + cur.execute('UPDATE contact_companies SET is_primary=FALSE WHERE customer_id=ANY(%s)',(sources,)) + for ref in movements: + table,column=ref['table_name'],ref['column_name'] + # Avoid introducing self-referential parent/billing relations on merged rows. + if table=='customers': + cur.execute(sql.SQL('UPDATE customers SET {}=NULL WHERE id=ANY(%s) AND {}=ANY(%s)').format( + sql.Identifier(column),sql.Identifier(column)),([target]+sources,sources)) + value=str(target) if table=='bmc_workflows' else target + ids=[str(i) for i in sources] if table=='bmc_workflows' else sources + cur.execute(sql.SQL('UPDATE {} SET {}=%s WHERE {}').format(sql.Identifier(table),sql.Identifier(column),where(ref)),(value,ids)) + cur.execute('UPDATE customers SET is_active=FALSE,merged_into_customer_id=%s,updated_at=now() WHERE id=ANY(%s)',(target,sources)) + return archived + + +def run(selection, actor_id, preview_token=None): + sources=selections(selection.target_id,selection.source_ids) + try: + with transaction(commit=preview_token is not None) as cur: + customers,changes,differences,movements,token,origins=prepare(cur,selection.target_id,sources) + if preview_token is not None and preview_token != token: + raise HTTPException(409,'Data er ændret siden forhåndsvisningen. Vis sammenlægningen igen.') + archived=move(cur,selection.target_id,sources,customers,changes,movements) + if preview_token is not None: + cur.execute('''INSERT INTO customer_merge_history(target_customer_id,source_customer_ids,actor_id,snapshot,movements) + VALUES(%s,%s,%s,%s,%s) RETURNING id''', + (selection.target_id,sources,actor_id,Json({'customers':customers,'relation_origins':origins,'archived_memberships':archived},dumps=lambda v:json.dumps(v,default=str)),Json(movements))) + history_id=cur.fetchone()['id'] + else: history_id=None + return {'target_id':selection.target_id,'source_ids':sources,'preview_token':token, + 'customers':[{'id':c['id'],'name':c['name']} for c in customers], + 'relations':[{'table':r['table_name'],'column':r['column_name'],'count':r['count']} for r in movements], + 'filled_fields':list(changes),'kept_main_fields':differences,'history_id':history_id, + 'external_systems_changed':False} + except psycopg2.IntegrityError as exc: + table=exc.diag.table_name or 'en relateret tabel' + raise HTTPException(409,f'Sammenlægningen har en konflikt i {table}. Ingen data er flyttet; afklar de overlappende relationer først.') from exc + except psycopg2.Error as exc: + logger.warning('Customer merge rolled back (%s)',exc.pgcode) + raise HTTPException(409,'Sammenlægningen kunne ikke gennemføres. Ingen data er flyttet. Genindlæs og prøv igen.') from exc + + +@router.post('/customers/merge/preview') +def preview(selection:MergeSelection,actor=Depends(require_permission('customers.edit'))): + return run(selection,actor['id']) + + +@router.post('/customers/merge') +def merge(selection:MergeConfirmation,actor=Depends(require_permission('customers.edit'))): + return run(selection,actor['id'],selection.preview_token) diff --git a/app/customers/backend/router.py b/app/customers/backend/router.py index 2effaf6..8bba840 100644 --- a/app/customers/backend/router.py +++ b/app/customers/backend/router.py @@ -23,6 +23,10 @@ logger = logging.getLogger(__name__) router = APIRouter() +# Static merge routes precede the legacy /customers/{customer_id} routes. +from app.customers.backend.merge import router as customer_merge_router +router.include_router(customer_merge_router) + def _ensure_customer_supplier_tag(customer_id: int) -> None: """Ensure linked customers are tagged as suppliers.""" diff --git a/app/customers/frontend/customer_detail.html b/app/customers/frontend/customer_detail.html index 4377fc8..ff62896 100644 --- a/app/customers/frontend/customer_detail.html +++ b/app/customers/frontend/customer_detail.html @@ -56,6 +56,26 @@ width: 20px; margin-right: 0.75rem; } + + /* Keep customer tabs isolated even when an external Bootstrap stylesheet is unavailable. */ + #customerTabContent > .tab-pane { + display: none; + } + + #customerTabContent > .tab-pane.active { + display: block; + opacity: 1; + } + + #customerTabLayout { + overflow-anchor: none; + } + + #customerTabNav .nav-link { + display: flex; + align-items: center; + min-height: 48px; + } .info-card { background: var(--bg-card); @@ -765,10 +785,10 @@ -
+
-
    +