feat(dispatch): ledger historique d assignation par tech (ops_tech_ticket_history)
Reconstruit la timeline complete d assignation des tickets depuis le CHANGE LOG
osTicket (ticket_msg: 'assign... change de X a Y') vers une table plate dans la
PG ERPNext (a cote de tabDispatch Job/tabIssue), incluant les tickets fermes.
- techHistoryBackfill(): scan watermarke borne de ticket_msg (I/O previsible),
parse + mapping nom->staff->TECH-<id>, upsert idempotent. Backfill initial + tick
incremental (un seul chemin) -> capte les nouveaux changements (les notres ET ceux de F).
- techHistory({tech}): rapport par tech (detail) ou recap tous techs.
- Routes: GET /dispatch/legacy-sync/tech-history[?tech=], GET|POST .../tech-history/backfill.
- Tick live planifie dans startSync (LEGACY_TECH_HISTORY, defaut on; LEGACY_TECH_HISTORY_MIN=15).
- Detecte 'de Tech Targo a <tech>' = sortie du pool vers un tech (from_pool).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
louispaulb
parent
e43f04d3e5
commit
6761c9e05a
|
|
@ -1154,6 +1154,7 @@ async function jobThread (jobName) {
|
|||
const _watchSnap = new Map() // ticket_id → { status, assign_to, last_update }
|
||||
let _watchTimer = null
|
||||
let _staleTimer = null
|
||||
let _techHistTimer = null
|
||||
let _lastWatch = null
|
||||
// Map staff legacy <id> → Dispatch Technician TECH-<id> (même logique que l'ingestion des tickets assignés).
|
||||
async function loadStaffToTech () {
|
||||
|
|
@ -1369,8 +1370,19 @@ function startSync () {
|
|||
_staleTimer = setInterval(stick, staleHours * 3600 * 1000)
|
||||
log(`stale-nudge: digest actif (toutes les ${staleHours}h)`)
|
||||
}
|
||||
// HISTORIQUE D'ASSIGNATION PAR TECH : tick incrémental (avance le watermark → capte les nouveaux changements).
|
||||
// Défaut ON (LEGACY_TECH_HISTORY=off pour couper). Petit lot/tick (le backfill initial complet = one-shot séparé).
|
||||
if (String(process.env.LEGACY_TECH_HISTORY || 'on').toLowerCase() !== 'off') {
|
||||
const thMin = Math.max(2, Number(process.env.LEGACY_TECH_HISTORY_MIN) || 15)
|
||||
const thBatches = Math.max(1, Number(process.env.LEGACY_TECH_HISTORY_TICK_BATCHES) || 8)
|
||||
let _thInFlight = false
|
||||
const thtick = () => { if (_thInFlight) return; _thInFlight = true; techHistoryBackfill({ maxBatches: thBatches }).then(r => { if (r && r.inserted) log('tech-history:', JSON.stringify({ inserted: r.inserted, matched: r.matched, wm: r.watermark })) }).catch(e => log('tech-history error:', e.message)).finally(() => { _thInFlight = false }) }
|
||||
setTimeout(thtick, 180 * 1000)
|
||||
_techHistTimer = setInterval(thtick, thMin * 60 * 1000)
|
||||
log(`tech-history: ledger d'assignation actif (toutes les ${thMin} min)`)
|
||||
}
|
||||
}
|
||||
function stopSync () { if (_timer) { clearInterval(_timer); _timer = null } if (_watchTimer) { clearInterval(_watchTimer); _watchTimer = null } if (_staleTimer) { clearInterval(_staleTimer); _staleTimer = null } }
|
||||
function stopSync () { if (_timer) { clearInterval(_timer); _timer = null } if (_watchTimer) { clearInterval(_watchTimer); _watchTimer = null } if (_staleTimer) { clearInterval(_staleTimer); _staleTimer = null } if (_techHistTimer) { clearInterval(_techHistTimer); _techHistTimer = null } }
|
||||
|
||||
// Jobs Legacy (osTicket) OUVERTS assignés à UN tech précis (staff_id legacy), filtrés « terrain »
|
||||
// (dépts dispatch : install/réparation/télé/téléphonie/monteur/fusion/désinstall OU subject « est. time »).
|
||||
|
|
@ -1975,6 +1987,13 @@ async function handle (req, res, method, path) {
|
|||
const u = new URL(req.url, 'http://localhost'); const jb = u.searchParams.get('job'); if (!jb) return json(res, 400, { ok: false, error: 'job requis' })
|
||||
return json(res, 200, await publishJob({ job: jb, dryRun: method === 'GET', actorEmail: req.headers['x-authentik-email'] || '', notify: u.searchParams.get('notify') !== '0', only: u.searchParams.get('only') || null }))
|
||||
}
|
||||
if (path === '/dispatch/legacy-sync/tech-history' && method === 'GET') { // ledger d'assignation par tech (PG). ?tech=TECH-<id> → détail · sans tech → récap par tech
|
||||
const u = new URL(req.url, 'http://localhost'); return json(res, 200, await techHistory({ tech: u.searchParams.get('tech') || '', limit: u.searchParams.get('limit') }))
|
||||
}
|
||||
if (path === '/dispatch/legacy-sync/tech-history/backfill' && (method === 'GET' || method === 'POST')) { // GET=dry-run · POST=applique. ?reset=1 repart de 0 · ?max=<batches> · ?batch=<taille>
|
||||
const u = new URL(req.url, 'http://localhost')
|
||||
return json(res, 200, await techHistoryBackfill({ dryRun: method === 'GET', reset: u.searchParams.get('reset') === '1', maxBatches: Number(u.searchParams.get('max')) || 40, batch: Number(u.searchParams.get('batch')) || 5000 }))
|
||||
}
|
||||
if (path === '/dispatch/legacy-sync/tech-sync' && method === 'GET') return json(res, 200, await techSyncReport()) // rapport réconciliation techs (3 systèmes)
|
||||
if (path === '/dispatch/legacy-sync/mine-durations' && method === 'GET') return json(res, 200, await mineDurations()) // baseline durées apprises (proxy réponses staff)
|
||||
if (path === '/dispatch/legacy-sync/tech-sync/apply' && method === 'POST') { // crée les fiches Dispatch Technician choisies (ERPNext only)
|
||||
|
|
@ -2487,4 +2506,87 @@ async function publishJob ({ job, dryRun = false, actorEmail = '', notify = true
|
|||
return { ok: applied.every(a => a.ok || a.skipped || a.dryRun), job, ticket, applied }
|
||||
}
|
||||
|
||||
module.exports = { handle, sync, reimportAddresses, fillMissingCoords, fixGeocoding, purgeStaleOrphans, pushAssignments, returnToPool, postTicketLegacy, ticketThread, watchLegacy, techSyncReport, techSyncApply, mineDurations, legacyTechTickets, legacyWindowLoad, ingestAssigned, reconcileLegacyJobs, backfillServiceLocations, backfillIssueCustomers, backfillSourceIssue, backfillDependsOn, startSync, stopSync, fetchTargoTickets, staleTickets, staleNudge, publishPreview, publishJob, coord, prio, startTime, jobType, inTerritory, geocodeRQA, persistGeocodeToSL, reverseNearestFibre } // parseurs purs exposés pour les tests
|
||||
// ═══ HISTORIQUE D'ASSIGNATION PAR TECH (ledger dans la PG ERPNext, à côté de tabDispatch Job/tabIssue) ═══
|
||||
// Source de vérité = `ticket_msg` CHANGE LOG « assign… change de <from> à <to> » (osTicket ne garde PAS de
|
||||
// table d'événements ; assign_to = l'assigné COURANT seulement). On reconstruit la timeline complète : chaque
|
||||
// passage d'un ticket d'un intervenant à un autre, avec date + qui a fait le changement + « venait du pool ».
|
||||
// UN SEUL chemin backfill+live : scan watermarké de ticket_msg (par id PK). Le 1er run backfille TOUT l'historique ;
|
||||
// un tick planifié avance le watermark → capte AUTO les nouveaux changements (les nôtres via ops_reassign ET ceux de F).
|
||||
const ASSIGN_RE = /assign\w*\s+change\s+de\s+(.+?)\s+à\s+([^.<\n]+)/i
|
||||
function techHistPg () { return require('./address-db').pool() }
|
||||
const TECH_HIST_WM = '/app/data/tech_history_watermark.json'
|
||||
async function techHistEnsureSchema () {
|
||||
const pg = techHistPg()
|
||||
await pg.query(`CREATE TABLE IF NOT EXISTS ops_tech_ticket_history (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
legacy_ticket_id BIGINT NOT NULL,
|
||||
changed_at TIMESTAMPTZ NOT NULL,
|
||||
to_name TEXT NOT NULL,
|
||||
to_staff_id INTEGER,
|
||||
tech_id TEXT,
|
||||
from_name TEXT,
|
||||
from_pool BOOLEAN DEFAULT FALSE,
|
||||
changed_by_staff INTEGER,
|
||||
dept TEXT,
|
||||
source TEXT DEFAULT 'backfill',
|
||||
UNIQUE (legacy_ticket_id, changed_at, to_name))`)
|
||||
await pg.query('CREATE INDEX IF NOT EXISTS idx_otth_tech ON ops_tech_ticket_history (tech_id)')
|
||||
await pg.query('CREATE INDEX IF NOT EXISTS idx_otth_ticket ON ops_tech_ticket_history (legacy_ticket_id)')
|
||||
}
|
||||
// index NOM legacy (actifs ET inactifs — l'historique inclut d'ex-employés) → staff.id
|
||||
async function loadStaffNameIndex () {
|
||||
const p = pool(); const [rows] = await p.query("SELECT id, TRIM(CONCAT(COALESCE(first_name,''),' ',COALESCE(last_name,''))) nm FROM staff")
|
||||
const byName = new Map(); for (const r of rows) { const k = norm(r.nm); if (k) byName.set(k, r.id) }
|
||||
return byName
|
||||
}
|
||||
// Scan borné : lit exactement `batch` lignes ticket_msg > watermark (I/O prévisible, indépendant de la densité de matches),
|
||||
// parse les logs d'assignation, upsert dans la PG. maxBatches plafonne par appel ; reprise via watermark persisté.
|
||||
async function techHistoryBackfill ({ batch = 5000, maxBatches = 40, throttleMs = 80, reset = false, dryRun = false } = {}) {
|
||||
await techHistEnsureSchema()
|
||||
const p = pool(); const pg = techHistPg()
|
||||
let wm = 0; if (!reset) { try { wm = JSON.parse(fs.readFileSync(TECH_HIST_WM, 'utf8')).id || 0 } catch (e) {} }
|
||||
const nameIdx = await loadStaffNameIndex(); const staffToTech = await loadStaffToTech()
|
||||
const [depts] = await p.query('SELECT id, name FROM ticket_dept'); const deptName = {}; for (const d of depts) deptName[d.id] = d.name
|
||||
let scanned = 0, matched = 0, inserted = 0, batches = 0
|
||||
for (; batches < maxBatches; batches++) {
|
||||
const [rows] = await p.query('SELECT tm.id, tm.ticket_id, tm.staff_id, tm.date_orig, tm.msg, t.dept_id FROM ticket_msg tm LEFT JOIN ticket t ON t.id = tm.ticket_id WHERE tm.id > ? ORDER BY tm.id ASC LIMIT ?', [wm, batch])
|
||||
if (!rows.length) break
|
||||
for (const m of rows) {
|
||||
wm = m.id; scanned++
|
||||
const mm = decodeEntities(m.msg).match(ASSIGN_RE); if (!mm) continue
|
||||
const fromName = mm[1].trim()
|
||||
let toName = mm[2].split(/\s+(?:date|CHANGE\s*LOG|statut|status|sujet|priorit)\b/i)[0].trim().slice(0, 120)
|
||||
if (!toName) continue
|
||||
matched++
|
||||
const toStaff = nameIdx.get(norm(toName)) || null
|
||||
const techId = toStaff ? (staffToTech[toStaff] || ('TECH-' + toStaff)) : null
|
||||
if (dryRun) continue
|
||||
const r = await pg.query(
|
||||
`INSERT INTO ops_tech_ticket_history (legacy_ticket_id, changed_at, to_name, to_staff_id, tech_id, from_name, from_pool, changed_by_staff, dept, source)
|
||||
VALUES ($1, to_timestamp($2), $3, $4, $5, $6, $7, $8, $9, 'backfill') ON CONFLICT (legacy_ticket_id, changed_at, to_name) DO NOTHING`,
|
||||
[m.ticket_id, m.date_orig, toName, toStaff, techId, fromName, /tech targo/i.test(fromName), m.staff_id, deptName[m.dept_id] || null])
|
||||
inserted += r.rowCount || 0
|
||||
}
|
||||
if (!dryRun) { try { fs.writeFileSync(TECH_HIST_WM, JSON.stringify({ id: wm, at: new Date().toISOString() })) } catch (e) {} }
|
||||
if (rows.length < batch) { batches++; break } // fin atteinte
|
||||
if (throttleMs) await new Promise(r => setTimeout(r, throttleMs))
|
||||
}
|
||||
return { ok: true, dryRun, scanned, matched, inserted, batches, watermark: wm, done: batches < maxBatches }
|
||||
}
|
||||
// Rapport : ?tech=TECH-<id> → lignes de ce tech ; sinon → récap par tech (compte + venant du pool + dernier).
|
||||
async function techHistory ({ tech = '', limit = 200 } = {}) {
|
||||
await techHistEnsureSchema()
|
||||
const pg = techHistPg(); const lim = Math.max(1, Math.min(2000, Number(limit) || 200))
|
||||
if (tech) {
|
||||
const r = await pg.query('SELECT legacy_ticket_id, changed_at, to_name, tech_id, from_name, from_pool, dept, changed_by_staff FROM ops_tech_ticket_history WHERE tech_id = $1 OR to_name = $1 ORDER BY changed_at DESC LIMIT $2', [tech, lim])
|
||||
const tot = await pg.query('SELECT COUNT(*)::int n, SUM(CASE WHEN from_pool THEN 1 ELSE 0 END)::int pool FROM ops_tech_ticket_history WHERE tech_id = $1 OR to_name = $1', [tech])
|
||||
return { ok: true, tech, total: tot.rows[0].n, from_pool: tot.rows[0].pool, rows: r.rows }
|
||||
}
|
||||
const r = await pg.query(`SELECT COALESCE(tech_id, to_name) key, MAX(to_name) name, COUNT(*)::int total,
|
||||
SUM(CASE WHEN from_pool THEN 1 ELSE 0 END)::int from_pool, MAX(changed_at) last_at
|
||||
FROM ops_tech_ticket_history GROUP BY COALESCE(tech_id, to_name) ORDER BY total DESC LIMIT $1`, [lim])
|
||||
const grand = await pg.query('SELECT COUNT(*)::int n, COUNT(DISTINCT COALESCE(tech_id,to_name))::int techs FROM ops_tech_ticket_history')
|
||||
return { ok: true, total_rows: grand.rows[0].n, techs: grand.rows[0].techs, per_tech: r.rows }
|
||||
}
|
||||
|
||||
module.exports = { handle, sync, reimportAddresses, fillMissingCoords, fixGeocoding, purgeStaleOrphans, pushAssignments, returnToPool, postTicketLegacy, ticketThread, watchLegacy, techSyncReport, techSyncApply, mineDurations, legacyTechTickets, legacyWindowLoad, ingestAssigned, reconcileLegacyJobs, backfillServiceLocations, backfillIssueCustomers, backfillSourceIssue, backfillDependsOn, startSync, stopSync, fetchTargoTickets, staleTickets, staleNudge, publishPreview, publishJob, techHistoryBackfill, techHistory, coord, prio, startTime, jobType, inTerritory, geocodeRQA, persistGeocodeToSL, reverseNearestFibre } // parseurs purs exposés pour les tests
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user