mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 02:00:46 +08:00
Add offline biz_state log import with unified create-task entry.
Create tasks via pick-NE or import log; re-import appends batches like collects, without inventory NE or SSH. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
7b5f03ad54
commit
d701e4a35e
13 changed files with 1991 additions and 139 deletions
|
|
@ -39,6 +39,13 @@ def enqueue_collect(task_id: str, *, manual: bool = False) -> dict[str, Any]:
|
|||
task = db.get(BizStateTask, tid)
|
||||
if not task:
|
||||
return {"ok": False, "queued": False, "reason": "task_not_found", "task_id": tid}
|
||||
if str(task.source or "").strip().lower() == "import":
|
||||
return {
|
||||
"ok": False,
|
||||
"queued": False,
|
||||
"reason": "import_offline_only",
|
||||
"task_id": tid,
|
||||
}
|
||||
if bool(task.collect_running):
|
||||
return {
|
||||
"ok": True,
|
||||
|
|
|
|||
662
netx_api/biz_state/import_runner.py
Normal file
662
netx_api/biz_state/import_runner.py
Normal file
|
|
@ -0,0 +1,662 @@
|
|||
"""Offline log import → same biz_state batch / parse / persist path (no SSH)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
from ..db import SessionLocal
|
||||
from ..lldp_shared import resolve_vendor_key
|
||||
from ..models import BizStateBatch, BizStateTask
|
||||
from ..timeutil import utcnow_naive
|
||||
from .collect_runner import (
|
||||
_finish_task,
|
||||
_finalize_batch_status,
|
||||
_format_error,
|
||||
_run_primary_parse_job,
|
||||
)
|
||||
from .collect_session import resolve_aux_command
|
||||
from .command_match import match_command, normalize_command
|
||||
from .log_split import LogSegment, unpack_upload
|
||||
from .parse_pool import AuxRawCapture, PrimaryParseJob
|
||||
from .parsers import get_parser
|
||||
from .persist_pool import get_persist_pool
|
||||
from .spool import (
|
||||
SpooledCommand,
|
||||
clear_batch_spool,
|
||||
count_text_lines,
|
||||
persist_every_cmds,
|
||||
write_meta,
|
||||
write_raw_text,
|
||||
)
|
||||
|
||||
_log = logging.getLogger("netx.biz_state.import")
|
||||
|
||||
|
||||
def _utcnow():
|
||||
return utcnow_naive()
|
||||
|
||||
|
||||
def _vendor_fields_for_key(vendor_key: str) -> tuple[str, str]:
|
||||
"""Map vendor_key → (vendor label, device_type) for offline import tasks."""
|
||||
key = str(vendor_key or "").strip().lower() or "zte"
|
||||
if key.startswith("huawei") or key in ("vrp", "ce", "ne"):
|
||||
return "Huawei", "huawei_vrp"
|
||||
if key.startswith("cisco") or key in ("ios", "nxos", "iosxe", "iosxr"):
|
||||
return "Cisco", "cisco_ios"
|
||||
if key.startswith("zte") or key in ("zxros", "zxr10"):
|
||||
return "ZTE", "zte_zxros"
|
||||
if key in ("generic", "any", "*"):
|
||||
return "generic", ""
|
||||
return key[:64] or "generic", ""
|
||||
|
||||
|
||||
def create_standalone_import_task(
|
||||
*,
|
||||
vendor_key: str,
|
||||
ne_name: str = "",
|
||||
note: str = "",
|
||||
filename: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Create a paused offline task (source=import) with no inventory NE."""
|
||||
vk = str(vendor_key or "").strip().lower() or "zte"
|
||||
vendor, device_type = _vendor_fields_for_key(vk)
|
||||
fname = Path(str(filename or "")).name[:120]
|
||||
label = str(ne_name or "").strip() or (fname and f"import:{fname}") or "offline-import"
|
||||
tid = uuid4().hex
|
||||
ne_id = f"import-{tid[:12]}"
|
||||
now = _utcnow()
|
||||
db = SessionLocal()
|
||||
try:
|
||||
task = BizStateTask(
|
||||
id=tid,
|
||||
source="import",
|
||||
ne_id=ne_id,
|
||||
ne_name=label[:256],
|
||||
ne_ip="",
|
||||
vendor=vendor,
|
||||
device_type=device_type,
|
||||
note=str(note or "")[:256],
|
||||
purpose="",
|
||||
status="paused", # offline only — no schedule / SSH
|
||||
interval_sec=3600,
|
||||
retention_days=30,
|
||||
daily_keep_enabled=False,
|
||||
daily_keep_count=10,
|
||||
retention_batches=30,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
return {
|
||||
"ok": True,
|
||||
"task_id": tid,
|
||||
"ne_id": ne_id,
|
||||
"ne_name": task.ne_name,
|
||||
"vendor": vendor,
|
||||
"device_type": device_type,
|
||||
"vendor_key": vk,
|
||||
}
|
||||
except Exception:
|
||||
_log.exception("create_standalone_import_task failed")
|
||||
try:
|
||||
db.rollback()
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def enqueue_import(
|
||||
task_id: str,
|
||||
*,
|
||||
filename: str = "",
|
||||
vendor_key: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Create a running import batch; mark task collect_running (mutex with SSH collect)."""
|
||||
tid = str(task_id or "").strip()
|
||||
if not tid:
|
||||
return {"ok": False, "queued": False, "reason": "missing_task_id", "task_id": ""}
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
task = db.get(BizStateTask, tid)
|
||||
if not task:
|
||||
return {"ok": False, "queued": False, "reason": "task_not_found", "task_id": tid}
|
||||
if bool(task.collect_running):
|
||||
return {
|
||||
"ok": True,
|
||||
"queued": False,
|
||||
"reason": "already_collecting",
|
||||
"task_id": tid,
|
||||
}
|
||||
st = str(task.status or "").strip()
|
||||
if st in ("", "deleted"):
|
||||
return {"ok": False, "queued": False, "reason": "bad_status", "task_id": tid}
|
||||
|
||||
now = _utcnow()
|
||||
task.collect_running = True
|
||||
task.last_collect_started_at = now
|
||||
task.last_error = ""
|
||||
task.updated_at = now
|
||||
if hasattr(task, "collect_queued_at"):
|
||||
task.collect_queued_at = now
|
||||
|
||||
fname = Path(str(filename or "import.log")).name[:120] or "import.log"
|
||||
vk = str(vendor_key or "").strip().lower()
|
||||
if not vk:
|
||||
vk = resolve_vendor_key(task.vendor or "", task.device_type or "")
|
||||
|
||||
batch = BizStateBatch(
|
||||
id=uuid4().hex,
|
||||
task_id=task.id,
|
||||
source=task.source,
|
||||
ne_id=task.ne_id,
|
||||
ne_name=task.ne_name,
|
||||
vendor=task.vendor,
|
||||
status="running",
|
||||
started_at=now,
|
||||
message=f"importing:{fname}",
|
||||
alias=fname[:128],
|
||||
)
|
||||
db.add(batch)
|
||||
db.commit()
|
||||
return {
|
||||
"ok": True,
|
||||
"queued": True,
|
||||
"batch_id": batch.id,
|
||||
"task_id": tid,
|
||||
"vendor_key": vk,
|
||||
"filename": fname,
|
||||
"manual": True,
|
||||
"import": True,
|
||||
}
|
||||
except Exception:
|
||||
_log.exception("enqueue_import failed task=%s", tid)
|
||||
try:
|
||||
db.rollback()
|
||||
except Exception:
|
||||
pass
|
||||
return {"ok": False, "queued": False, "reason": "enqueue_failed", "task_id": tid}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def _segment_index(segments: list[LogSegment]) -> dict[str, LogSegment]:
|
||||
"""Map normalize_command(cmd) → first matching segment."""
|
||||
idx: dict[str, LogSegment] = {}
|
||||
for seg in segments:
|
||||
key = normalize_command(seg.command)
|
||||
if key and key not in idx:
|
||||
idx[key] = seg
|
||||
return idx
|
||||
|
||||
|
||||
def _build_aux_captures(
|
||||
*,
|
||||
profile: Any,
|
||||
params: dict[str, str],
|
||||
batch_id: str,
|
||||
seg_index: dict[str, LogSegment],
|
||||
) -> tuple[list[AuxRawCapture], list[str]]:
|
||||
"""Try to satisfy aux_commands from segments already in the import log."""
|
||||
captures: list[AuxRawCapture] = []
|
||||
notes: list[str] = []
|
||||
for aux in list(getattr(profile, "aux_commands", None) or []):
|
||||
try:
|
||||
ra = resolve_aux_command(aux, params)
|
||||
except Exception as exc:
|
||||
notes.append(f"aux:{getattr(aux, 'key', '?')}:resolve:{exc}")
|
||||
continue
|
||||
hit = seg_index.get(normalize_command(ra.command))
|
||||
if hit is None:
|
||||
notes.append(f"enrich_skipped:missing_aux:{ra.key}")
|
||||
continue
|
||||
aux_id = uuid4().hex
|
||||
raw_rel = ""
|
||||
try:
|
||||
raw_rel = write_raw_text(batch_id, aux_id, hit.body)
|
||||
except Exception:
|
||||
_log.exception("import aux spool raw failed")
|
||||
captures.append(
|
||||
AuxRawCapture(
|
||||
key=ra.key,
|
||||
aux_id=aux_id,
|
||||
profile_id=ra.profile_id,
|
||||
parser_id=ra.parser_id,
|
||||
metric_id=str(getattr(ra.profile, "metric_id", "") or ""),
|
||||
command=ra.command,
|
||||
textfsm_command=ra.textfsm_command or ra.command,
|
||||
rule_keys=tuple(ra.rule_keys or ()),
|
||||
raw=hit.body,
|
||||
raw_rel_path=raw_rel,
|
||||
cache_hit=False,
|
||||
ok=True,
|
||||
)
|
||||
)
|
||||
return captures, notes
|
||||
|
||||
|
||||
def run_import_batch(
|
||||
*,
|
||||
batch_id: str,
|
||||
task_id: str,
|
||||
segments: list[LogSegment],
|
||||
vendor_key: str,
|
||||
vendor: str = "",
|
||||
device_type: str = "",
|
||||
filename: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Parse imported segments into the batch (sync parse + persist pool)."""
|
||||
bid = str(batch_id or "").strip()
|
||||
tid = str(task_id or "").strip()
|
||||
vk = str(vendor_key or "").strip().lower() or "zte"
|
||||
any_ok = False
|
||||
any_fail = False
|
||||
matched = 0
|
||||
unmatched = 0
|
||||
cmd_count = 0
|
||||
lane_errors: list[str] = []
|
||||
|
||||
try:
|
||||
clear_batch_spool(bid)
|
||||
except Exception:
|
||||
_log.exception("import clear spool failed batch=%s", bid)
|
||||
|
||||
persist = get_persist_pool()
|
||||
pending: list[SpooledCommand] = []
|
||||
flush_every = persist_every_cmds()
|
||||
seg_index = _segment_index(segments)
|
||||
persisted: set[tuple[str, str]] = set()
|
||||
cache_lock = threading.RLock()
|
||||
|
||||
def _submit_pending() -> None:
|
||||
nonlocal pending
|
||||
if not pending:
|
||||
return
|
||||
chunk = list(pending)
|
||||
pending = []
|
||||
persist.submit(bid, chunk)
|
||||
|
||||
def _queue(item: SpooledCommand) -> None:
|
||||
nonlocal cmd_count
|
||||
try:
|
||||
write_meta(bid, item.id, item.to_meta())
|
||||
except Exception:
|
||||
_log.exception("import write meta failed cmd=%s", item.id)
|
||||
pending.append(item)
|
||||
cmd_count += 1
|
||||
if len(pending) >= flush_every:
|
||||
_submit_pending()
|
||||
|
||||
for seg in segments:
|
||||
cmd = normalize_command(seg.command)
|
||||
cmd_id = uuid4().hex
|
||||
src_note = ""
|
||||
if seg.source_file:
|
||||
src_note = f"file={seg.source_file}"
|
||||
hit = match_command(vendor_key=vk, command=cmd)
|
||||
raw_rel = ""
|
||||
try:
|
||||
raw_rel = write_raw_text(bid, cmd_id, seg.body)
|
||||
except Exception:
|
||||
_log.exception("import spool raw failed cmd=%s", cmd_id)
|
||||
|
||||
if not hit:
|
||||
unmatched += 1
|
||||
any_fail = True
|
||||
msg = "no profile matched concrete command"
|
||||
if src_note:
|
||||
msg = f"{msg};{src_note}"
|
||||
_queue(
|
||||
SpooledCommand(
|
||||
id=cmd_id,
|
||||
batch_id=bid,
|
||||
profile_id="",
|
||||
raw_command=cmd[:512],
|
||||
parse_status="unmatched",
|
||||
message=msg[:1020],
|
||||
raw_rel_path=raw_rel,
|
||||
raw_line_count=count_text_lines(seg.body),
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
if not get_parser(hit.profile.parser_id):
|
||||
unmatched += 1
|
||||
any_fail = True
|
||||
_queue(
|
||||
SpooledCommand(
|
||||
id=cmd_id,
|
||||
batch_id=bid,
|
||||
profile_id=hit.profile.profile_id,
|
||||
parser_id=hit.profile.parser_id,
|
||||
metric_id=hit.profile.metric_id,
|
||||
raw_command=cmd[:512],
|
||||
params_json=dict(hit.params or {}),
|
||||
parse_status="failed",
|
||||
message=f"unknown parser {hit.profile.parser_id}"[:1020],
|
||||
raw_rel_path=raw_rel,
|
||||
raw_line_count=count_text_lines(seg.body),
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
matched += 1
|
||||
aux_caps, aux_notes = _build_aux_captures(
|
||||
profile=hit.profile,
|
||||
params=dict(hit.params or {}),
|
||||
batch_id=bid,
|
||||
seg_index=seg_index,
|
||||
)
|
||||
enrich = list(hit.profile.enrich_joins or [])
|
||||
# Without in-log aux, skip enrich so we don't pretend peer intent joined.
|
||||
if any("enrich_skipped" in n for n in aux_notes):
|
||||
enrich = []
|
||||
job = PrimaryParseJob(
|
||||
batch_id=bid,
|
||||
cmd_id=cmd_id,
|
||||
task_item_id="",
|
||||
profile_id=hit.profile.profile_id,
|
||||
parser_id=hit.profile.parser_id,
|
||||
metric_id=hit.profile.metric_id,
|
||||
concrete=cmd,
|
||||
merged_params=dict(hit.params or {}),
|
||||
raw_text=seg.body,
|
||||
raw_rel_path=raw_rel,
|
||||
raw_line_count=count_text_lines(seg.body),
|
||||
textfsm_command=hit.profile.textfsm_command or cmd,
|
||||
vendor=vendor,
|
||||
device_type=device_type,
|
||||
enrich_joins=enrich,
|
||||
aux_captures=aux_caps,
|
||||
persisted=persisted,
|
||||
cache_lock=cache_lock,
|
||||
on_done=None,
|
||||
)
|
||||
try:
|
||||
ok, fail = _run_primary_parse_job(job)
|
||||
if ok:
|
||||
any_ok = True
|
||||
if fail:
|
||||
any_fail = True
|
||||
if aux_notes:
|
||||
_log.info(
|
||||
"import enrich notes batch=%s cmd=%s %s",
|
||||
bid,
|
||||
cmd_id,
|
||||
";".join(aux_notes),
|
||||
)
|
||||
cmd_count += 1 + len(aux_caps)
|
||||
except Exception as exc:
|
||||
any_fail = True
|
||||
_queue(
|
||||
SpooledCommand(
|
||||
id=cmd_id,
|
||||
batch_id=bid,
|
||||
profile_id=hit.profile.profile_id,
|
||||
parser_id=hit.profile.parser_id,
|
||||
metric_id=hit.profile.metric_id,
|
||||
raw_command=cmd[:512],
|
||||
params_json=dict(hit.params or {}),
|
||||
parse_status="failed",
|
||||
message=f"parse: {_format_error(exc)}"[:1020],
|
||||
raw_rel_path=raw_rel,
|
||||
raw_line_count=count_text_lines(seg.body),
|
||||
)
|
||||
)
|
||||
|
||||
_submit_pending()
|
||||
# _run_primary_parse_job also submits to this same pool — wait once for all.
|
||||
if not persist.wait_idle(timeout=3600.0):
|
||||
any_fail = True
|
||||
lane_errors.append("import: persist_barrier_timeout")
|
||||
|
||||
fname = Path(str(filename or "")).name
|
||||
summary = (
|
||||
f"imported:{fname or 'upload'}; segments={len(segments)}; "
|
||||
f"matched={matched}; unmatched={unmatched}"
|
||||
)
|
||||
if fname:
|
||||
lane_errors.insert(0, summary)
|
||||
|
||||
status = ""
|
||||
try:
|
||||
status = _finalize_batch_status(
|
||||
batch_id=bid,
|
||||
task_id=tid,
|
||||
cmd_count=cmd_count,
|
||||
total_rows=0,
|
||||
any_fail=any_fail,
|
||||
any_ok=any_ok or matched > 0,
|
||||
lane_errors=lane_errors,
|
||||
)
|
||||
except Exception as exc:
|
||||
_log.exception("import finalize failed batch=%s", bid)
|
||||
lane_errors.append(_format_error(exc))
|
||||
try:
|
||||
from .collect_runner import _fail_batch_status
|
||||
|
||||
_fail_batch_status(bid, "; ".join(lane_errors)[:1020])
|
||||
except Exception:
|
||||
pass
|
||||
status = "failed"
|
||||
|
||||
# Full success clears message in finalize — keep import provenance visible.
|
||||
if status == "success":
|
||||
try:
|
||||
_stamp_import_message(bid, summary)
|
||||
except Exception:
|
||||
_log.exception("import stamp message failed batch=%s", bid)
|
||||
|
||||
err = ""
|
||||
if status in ("failed",) and not any_ok:
|
||||
err = "; ".join(lane_errors)[:1020] or "import failed"
|
||||
try:
|
||||
_finish_task(tid, error=err)
|
||||
except Exception:
|
||||
_log.exception("import finish task failed task=%s", tid)
|
||||
|
||||
try:
|
||||
clear_batch_spool(bid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return {
|
||||
"ok": status in ("success", "partial"),
|
||||
"batch_id": bid,
|
||||
"task_id": tid,
|
||||
"status": status,
|
||||
"segments": len(segments),
|
||||
"matched": matched,
|
||||
"unmatched": unmatched,
|
||||
"command_count": cmd_count,
|
||||
}
|
||||
|
||||
|
||||
def _stamp_import_message(batch_id: str, summary: str) -> None:
|
||||
db = SessionLocal()
|
||||
try:
|
||||
batch = db.get(BizStateBatch, batch_id)
|
||||
if not batch:
|
||||
return
|
||||
if not str(batch.message or "").strip():
|
||||
batch.message = str(summary or "")[:1020]
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def start_standalone_import(
|
||||
*,
|
||||
filename: str,
|
||||
data: bytes,
|
||||
vendor_key: str = "",
|
||||
ne_name: str = "",
|
||||
note: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Create offline task + unpack + enqueue import (no inventory NE)."""
|
||||
vk = str(vendor_key or "").strip().lower() or "zte"
|
||||
try:
|
||||
segments, stats = unpack_upload(
|
||||
filename=filename, data=data, vendor_key=vk
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
if not segments:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="no show/display command segments found in upload",
|
||||
)
|
||||
|
||||
created = create_standalone_import_task(
|
||||
vendor_key=vk,
|
||||
ne_name=ne_name,
|
||||
note=note,
|
||||
filename=filename,
|
||||
)
|
||||
tid = str(created["task_id"])
|
||||
vendor = str(created.get("vendor") or "")
|
||||
device_type = str(created.get("device_type") or "")
|
||||
|
||||
result = enqueue_import(tid, filename=filename, vendor_key=vk)
|
||||
if not result.get("queued"):
|
||||
# Best-effort: leave the empty task for the user to retry / delete.
|
||||
return {
|
||||
"ok": bool(result.get("ok", False)),
|
||||
"started": False,
|
||||
"queued": False,
|
||||
"reason": result.get("reason") or "enqueue_failed",
|
||||
"task_id": tid,
|
||||
"created_task": True,
|
||||
**stats,
|
||||
}
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"started": True,
|
||||
"queued": True,
|
||||
"batch_id": result["batch_id"],
|
||||
"task_id": tid,
|
||||
"vendor_key": vk,
|
||||
"filename": result.get("filename") or filename,
|
||||
"vendor": vendor,
|
||||
"device_type": device_type,
|
||||
"ne_name": created.get("ne_name") or "",
|
||||
"created_task": True,
|
||||
"segments_preview": stats,
|
||||
"segments": segments,
|
||||
"collect_running": True,
|
||||
}
|
||||
|
||||
|
||||
def start_import_from_upload(
|
||||
*,
|
||||
task_id: str,
|
||||
filename: str,
|
||||
data: bytes,
|
||||
vendor_key: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Enqueue + unpack; caller should run ``execute_import`` in background."""
|
||||
tid = str(task_id or "").strip()
|
||||
db = SessionLocal()
|
||||
try:
|
||||
task = db.get(BizStateTask, tid)
|
||||
if not task:
|
||||
raise HTTPException(status_code=404, detail="task not found")
|
||||
vendor = str(task.vendor or "")
|
||||
device_type = str(task.device_type or "")
|
||||
vk = str(vendor_key or "").strip().lower()
|
||||
if not vk:
|
||||
vk = resolve_vendor_key(vendor, device_type)
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
try:
|
||||
segments, stats = unpack_upload(
|
||||
filename=filename, data=data, vendor_key=vk
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
if not segments:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="no show/display command segments found in upload",
|
||||
)
|
||||
|
||||
result = enqueue_import(tid, filename=filename, vendor_key=vk)
|
||||
if not result.get("queued"):
|
||||
return {
|
||||
"ok": bool(result.get("ok", False)),
|
||||
"started": False,
|
||||
"queued": False,
|
||||
"reason": result.get("reason") or "enqueue_failed",
|
||||
"task_id": tid,
|
||||
**stats,
|
||||
}
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"started": True,
|
||||
"queued": True,
|
||||
"batch_id": result["batch_id"],
|
||||
"task_id": tid,
|
||||
"vendor_key": vk,
|
||||
"filename": result.get("filename") or filename,
|
||||
"vendor": vendor,
|
||||
"device_type": device_type,
|
||||
"segments_preview": stats,
|
||||
"segments": segments, # passed to background runner (in-memory)
|
||||
"collect_running": True,
|
||||
}
|
||||
|
||||
|
||||
def execute_import(
|
||||
*,
|
||||
batch_id: str,
|
||||
task_id: str,
|
||||
segments: list[LogSegment],
|
||||
vendor_key: str,
|
||||
vendor: str = "",
|
||||
device_type: str = "",
|
||||
filename: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Background entry: parse imported segments."""
|
||||
try:
|
||||
return run_import_batch(
|
||||
batch_id=batch_id,
|
||||
task_id=task_id,
|
||||
segments=segments,
|
||||
vendor_key=vendor_key,
|
||||
vendor=vendor,
|
||||
device_type=device_type,
|
||||
filename=filename,
|
||||
)
|
||||
except Exception as exc:
|
||||
_log.exception("execute_import failed batch=%s", batch_id)
|
||||
try:
|
||||
from .collect_runner import _fail_batch_status
|
||||
|
||||
_fail_batch_status(batch_id, _format_error(exc))
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
_finish_task(task_id, error=_format_error(exc))
|
||||
except Exception:
|
||||
pass
|
||||
return {
|
||||
"ok": False,
|
||||
"batch_id": batch_id,
|
||||
"task_id": task_id,
|
||||
"status": "failed",
|
||||
"error": _format_error(exc),
|
||||
}
|
||||
241
netx_api/biz_state/log_split.py
Normal file
241
netx_api/biz_state/log_split.py
Normal file
|
|
@ -0,0 +1,241 @@
|
|||
"""Split device CLI transcript logs into show/display command segments."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import re
|
||||
import zipfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import BinaryIO
|
||||
|
||||
from ..config import settings
|
||||
|
||||
# Text-like extensions accepted from zip members / bare uploads.
|
||||
_TEXT_SUFFIXES = {".txt", ".log", ".ini", ".cfg", ".cli", ".out", ".text"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LogSegment:
|
||||
"""One CLI command and its captured output body."""
|
||||
|
||||
command: str
|
||||
body: str
|
||||
source_file: str = ""
|
||||
line_start: int = 0
|
||||
|
||||
|
||||
def import_max_bytes() -> int:
|
||||
return max(
|
||||
1,
|
||||
int(getattr(settings, "biz_state_import_max_bytes", 256 * 1024 * 1024) or 0)
|
||||
or (256 * 1024 * 1024),
|
||||
)
|
||||
|
||||
|
||||
def import_max_files() -> int:
|
||||
return max(1, int(getattr(settings, "biz_state_import_max_files", 200) or 200))
|
||||
|
||||
|
||||
def normalize_log_text(text: str) -> str:
|
||||
"""NBSP → space, unify newlines, strip trailing CR."""
|
||||
s = str(text or "").replace("\u00a0", " ").replace("\r\n", "\n").replace("\r", "\n")
|
||||
return s
|
||||
|
||||
|
||||
def _anchor_re(vendor_key: str) -> re.Pattern[str]:
|
||||
key = str(vendor_key or "").strip().lower()
|
||||
if key.startswith("huawei") or key in ("vrp", "ce", "ne"):
|
||||
# Prefer display; also accept show (mixed dumps).
|
||||
return re.compile(r"(?im)^(?:display|show)\s+")
|
||||
if key.startswith("cisco") or key in ("ios", "nxos", "iosxe", "iosxr"):
|
||||
return re.compile(r"(?im)^(?:show)\s+")
|
||||
if key.startswith("zte") or key in ("zxros", "zxr10"):
|
||||
return re.compile(r"(?im)^(?:show)\s+")
|
||||
# Unknown / generic: both
|
||||
return re.compile(r"(?im)^(?:show|display)\s+")
|
||||
|
||||
|
||||
_PROMPT_LINE_RE = re.compile(r"^[A-Za-z0-9._\-\[\]/]+[#>]\s*(.+)$")
|
||||
_HW_PROMPT_LINE_RE = re.compile(r"^<[^>]+>\s*(.+)$")
|
||||
|
||||
|
||||
def _strip_prompt_noise(line: str) -> str:
|
||||
"""Drop hostname# / hostname> / <VRP> prefixes from a command line."""
|
||||
s = line.strip()
|
||||
if not s:
|
||||
return ""
|
||||
# Whole-line prompt alone
|
||||
if re.fullmatch(r"[A-Za-z0-9._\-\[\]/]+[#>]", s):
|
||||
return ""
|
||||
if re.fullmatch(r"<[^>]+>", s):
|
||||
return ""
|
||||
# "R1#show arp" → "show arp"
|
||||
m = _PROMPT_LINE_RE.match(s)
|
||||
if m:
|
||||
return m.group(1).strip()
|
||||
# "<HUAWEI>display ip routing-table"
|
||||
m = _HW_PROMPT_LINE_RE.match(s)
|
||||
if m:
|
||||
return m.group(1).strip()
|
||||
return s
|
||||
|
||||
|
||||
def split_log_text(
|
||||
text: str,
|
||||
*,
|
||||
vendor_key: str = "",
|
||||
source_file: str = "",
|
||||
) -> list[LogSegment]:
|
||||
"""Split a CLI transcript into show/display segments.
|
||||
|
||||
Handles real device pastes such as::
|
||||
|
||||
MDN-BCP-CN1-ZM8SP#show arp | one-line
|
||||
...
|
||||
MDN-BCP-CN1-ZM8SP#show interface brief
|
||||
...
|
||||
|
||||
Prompt prefixes (``host#`` / ``host>`` / ``<VRP>``) are stripped before
|
||||
matching; each new show/display line starts a new segment.
|
||||
|
||||
Anything before the first show/display (banners, clocks, lone prompts) is
|
||||
discarded. If the whole text has no show/display, returns an empty list.
|
||||
"""
|
||||
raw = normalize_log_text(text)
|
||||
if not raw.strip():
|
||||
return []
|
||||
anchor = _anchor_re(vendor_key)
|
||||
lines = raw.split("\n")
|
||||
starts: list[tuple[int, str]] = [] # (0-based line idx, command)
|
||||
for i, line in enumerate(lines):
|
||||
cleaned = _strip_prompt_noise(line)
|
||||
if not cleaned:
|
||||
continue
|
||||
if anchor.match(cleaned):
|
||||
# Command is the cleaned line (may include | filters)
|
||||
cmd = re.sub(r"\s+", " ", cleaned).strip()
|
||||
starts.append((i, cmd))
|
||||
|
||||
if not starts:
|
||||
return []
|
||||
|
||||
out: list[LogSegment] = []
|
||||
for idx, (line_i, cmd) in enumerate(starts):
|
||||
end = starts[idx + 1][0] if idx + 1 < len(starts) else len(lines)
|
||||
body_lines = lines[line_i + 1 : end]
|
||||
# Drop leading blank lines; keep rest (incl. prompts inside body — parsers skip them)
|
||||
while body_lines and not body_lines[0].strip():
|
||||
body_lines = body_lines[1:]
|
||||
# Trim trailing blank
|
||||
while body_lines and not body_lines[-1].strip():
|
||||
body_lines = body_lines[:-1]
|
||||
body = "\n".join(body_lines)
|
||||
out.append(
|
||||
LogSegment(
|
||||
command=cmd,
|
||||
body=body,
|
||||
source_file=str(source_file or ""),
|
||||
line_start=line_i + 1,
|
||||
)
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def _decode_bytes(data: bytes) -> str:
|
||||
for enc in ("utf-8", "utf-8-sig", "gb18030", "latin-1"):
|
||||
try:
|
||||
return data.decode(enc)
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
return data.decode("utf-8", errors="replace")
|
||||
|
||||
|
||||
def _is_text_member(name: str) -> bool:
|
||||
n = str(name or "").replace("\\", "/").strip()
|
||||
if not n or n.endswith("/"):
|
||||
return False
|
||||
base = Path(n).name
|
||||
if base.startswith(".") or base.startswith("__MACOSX"):
|
||||
return False
|
||||
suf = Path(base).suffix.lower()
|
||||
if suf in _TEXT_SUFFIXES:
|
||||
return True
|
||||
# Extensionless small dumps sometimes appear; allow if no suffix
|
||||
return suf == ""
|
||||
|
||||
|
||||
def unpack_upload(
|
||||
*,
|
||||
filename: str,
|
||||
data: bytes,
|
||||
vendor_key: str = "",
|
||||
) -> tuple[list[LogSegment], dict[str, int]]:
|
||||
"""Unpack a bare text upload or zip into ordered LogSegments.
|
||||
|
||||
Returns (segments, stats) where stats has files / bytes / segments counts.
|
||||
Raises ValueError on size / format / zip-bomb limits.
|
||||
"""
|
||||
name = str(filename or "upload.bin").strip() or "upload.bin"
|
||||
blob = data or b""
|
||||
max_b = import_max_bytes()
|
||||
max_f = import_max_files()
|
||||
if len(blob) > max_b:
|
||||
raise ValueError(f"upload exceeds max size {max_b}B")
|
||||
|
||||
lower = name.lower()
|
||||
segments: list[LogSegment] = []
|
||||
total_bytes = 0
|
||||
file_count = 0
|
||||
|
||||
if lower.endswith(".zip"):
|
||||
try:
|
||||
zf = zipfile.ZipFile(io.BytesIO(blob))
|
||||
except zipfile.BadZipFile as exc:
|
||||
raise ValueError(f"invalid zip: {exc}") from exc
|
||||
with zf:
|
||||
members = [
|
||||
info
|
||||
for info in zf.infolist()
|
||||
if not info.is_dir() and _is_text_member(info.filename)
|
||||
]
|
||||
if len(members) > max_f:
|
||||
raise ValueError(f"zip has too many text files (>{max_f})")
|
||||
for info in sorted(members, key=lambda x: x.filename.lower()):
|
||||
if info.file_size > max_b:
|
||||
raise ValueError(
|
||||
f"zip member {info.filename!r} exceeds max size {max_b}B"
|
||||
)
|
||||
# Zip bomb: compressed ratio / total uncompressed
|
||||
total_bytes += int(info.file_size or 0)
|
||||
if total_bytes > max_b:
|
||||
raise ValueError(f"zip uncompressed total exceeds max size {max_b}B")
|
||||
raw = zf.read(info)
|
||||
text = _decode_bytes(raw)
|
||||
file_count += 1
|
||||
segs = split_log_text(
|
||||
text, vendor_key=vendor_key, source_file=info.filename
|
||||
)
|
||||
segments.extend(segs)
|
||||
else:
|
||||
total_bytes = len(blob)
|
||||
text = _decode_bytes(blob)
|
||||
file_count = 1
|
||||
segments = split_log_text(text, vendor_key=vendor_key, source_file=name)
|
||||
|
||||
return segments, {
|
||||
"files": file_count,
|
||||
"bytes": total_bytes,
|
||||
"segments": len(segments),
|
||||
}
|
||||
|
||||
|
||||
def read_upload_stream(fh: BinaryIO, *, max_bytes: int | None = None) -> bytes:
|
||||
"""Read upload stream with a hard byte cap."""
|
||||
cap = int(max_bytes if max_bytes is not None else import_max_bytes())
|
||||
buf = fh.read(cap + 1)
|
||||
if buf is None:
|
||||
return b""
|
||||
if len(buf) > cap:
|
||||
raise ValueError(f"upload exceeds max size {cap}B")
|
||||
return buf
|
||||
|
|
@ -251,6 +251,10 @@ def update_task(db: Session, task_id: str, body: dict[str, Any]) -> dict[str, An
|
|||
task = db.get(BizStateTask, task_id)
|
||||
if not task:
|
||||
raise HTTPException(status_code=404, detail="task_not_found")
|
||||
# Offline import tasks have no inventory NE / credentials — keep paused.
|
||||
if str(task.source or "").strip().lower() == "import":
|
||||
if "status" in body and str(body.get("status") or "").strip() == "running":
|
||||
raise HTTPException(status_code=400, detail="import_offline_only")
|
||||
if "note" in body:
|
||||
task.note = str(body.get("note") or "")[:256]
|
||||
if "purpose" in body and body["purpose"] is not None:
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ from __future__ import annotations
|
|||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, BackgroundTasks, Depends, Query
|
||||
from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, Query, UploadFile
|
||||
from fastapi.responses import StreamingResponse
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy.orm import Session
|
||||
|
|
@ -287,6 +287,153 @@ def api_collect_stop(task_id: str, db: Session = Depends(get_db)) -> dict[str, A
|
|||
return request_stop_collect(task_id)
|
||||
|
||||
|
||||
@router.post("/import")
|
||||
async def api_import_log_standalone(
|
||||
background_tasks: BackgroundTasks,
|
||||
file: UploadFile = File(...),
|
||||
vendor_key: str = Form("zte"),
|
||||
ne_name: str = Form(""),
|
||||
note: str = Form(""),
|
||||
) -> dict[str, Any]:
|
||||
"""Standalone offline import: create an import task (no NE) and parse the log."""
|
||||
from fastapi import HTTPException
|
||||
|
||||
from .biz_state.import_runner import execute_import, start_standalone_import
|
||||
from .biz_state.log_split import import_max_bytes, read_upload_stream
|
||||
|
||||
fname = str(file.filename or "import.log")
|
||||
try:
|
||||
data = read_upload_stream(file.file, max_bytes=import_max_bytes())
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
result = start_standalone_import(
|
||||
filename=fname,
|
||||
data=data,
|
||||
vendor_key=str(vendor_key or "").strip() or "zte",
|
||||
ne_name=str(ne_name or "").strip(),
|
||||
note=str(note or "").strip(),
|
||||
)
|
||||
if not result.get("queued"):
|
||||
return {
|
||||
"ok": bool(result.get("ok", False)),
|
||||
"started": False,
|
||||
"queued": False,
|
||||
"reason": result.get("reason") or "enqueue_failed",
|
||||
"task_id": result.get("task_id") or "",
|
||||
"created_task": bool(result.get("created_task")),
|
||||
"segments_preview": result.get("segments_preview")
|
||||
or {"files": result.get("files"), "segments": result.get("segments")},
|
||||
}
|
||||
|
||||
segments = result.pop("segments", [])
|
||||
bid = str(result.get("batch_id") or "")
|
||||
tid = str(result.get("task_id") or "")
|
||||
vk = str(result.get("vendor_key") or "zte")
|
||||
vendor = str(result.get("vendor") or "")
|
||||
device_type = str(result.get("device_type") or "")
|
||||
background_tasks.add_task(
|
||||
lambda: execute_import(
|
||||
batch_id=bid,
|
||||
task_id=tid,
|
||||
segments=list(segments),
|
||||
vendor_key=vk,
|
||||
vendor=vendor,
|
||||
device_type=device_type,
|
||||
filename=fname,
|
||||
)
|
||||
)
|
||||
return {
|
||||
"ok": True,
|
||||
"started": True,
|
||||
"queued": True,
|
||||
"batch_id": bid,
|
||||
"task_id": tid,
|
||||
"vendor_key": vk,
|
||||
"filename": result.get("filename") or fname,
|
||||
"ne_name": result.get("ne_name") or "",
|
||||
"created_task": True,
|
||||
"segments_preview": result.get("segments_preview") or {},
|
||||
"collect_running": True,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/tasks/{task_id}/import")
|
||||
async def api_import_log(
|
||||
task_id: str,
|
||||
background_tasks: BackgroundTasks,
|
||||
file: UploadFile = File(...),
|
||||
vendor_key: str = Form(""),
|
||||
db: Session = Depends(get_db),
|
||||
) -> dict[str, Any]:
|
||||
"""Upload a CLI transcript into an existing task and parse offline."""
|
||||
from fastapi import HTTPException
|
||||
|
||||
from .biz_state.import_runner import execute_import, start_import_from_upload
|
||||
from .biz_state.log_split import import_max_bytes, read_upload_stream
|
||||
|
||||
task = db.get(BizStateTask, task_id)
|
||||
if not task:
|
||||
raise HTTPException(status_code=404, detail="task not found")
|
||||
if bool(task.collect_running):
|
||||
return {
|
||||
"ok": True,
|
||||
"started": False,
|
||||
"reason": "already_collecting",
|
||||
"task_id": task_id,
|
||||
}
|
||||
|
||||
fname = str(file.filename or "import.log")
|
||||
try:
|
||||
data = read_upload_stream(file.file, max_bytes=import_max_bytes())
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
result = start_import_from_upload(
|
||||
task_id=task_id,
|
||||
filename=fname,
|
||||
data=data,
|
||||
vendor_key=str(vendor_key or "").strip(),
|
||||
)
|
||||
if not result.get("queued"):
|
||||
return {
|
||||
"ok": bool(result.get("ok", False)),
|
||||
"started": False,
|
||||
"queued": False,
|
||||
"reason": result.get("reason") or "enqueue_failed",
|
||||
"task_id": task_id,
|
||||
"segments_preview": result.get("segments_preview") or result.get("files"),
|
||||
}
|
||||
|
||||
segments = result.pop("segments", [])
|
||||
bid = str(result.get("batch_id") or "")
|
||||
vk = str(result.get("vendor_key") or "")
|
||||
vendor = str(result.get("vendor") or "")
|
||||
device_type = str(result.get("device_type") or "")
|
||||
background_tasks.add_task(
|
||||
lambda: execute_import(
|
||||
batch_id=bid,
|
||||
task_id=task_id,
|
||||
segments=list(segments),
|
||||
vendor_key=vk,
|
||||
vendor=vendor,
|
||||
device_type=device_type,
|
||||
filename=fname,
|
||||
)
|
||||
)
|
||||
return {
|
||||
"ok": True,
|
||||
"started": True,
|
||||
"queued": True,
|
||||
"batch_id": bid,
|
||||
"task_id": task_id,
|
||||
"vendor_key": vk,
|
||||
"filename": result.get("filename") or fname,
|
||||
"segments_preview": result.get("segments_preview") or {},
|
||||
"collect_running": True,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/tasks/{task_id}/batches")
|
||||
def api_list_batches(
|
||||
task_id: str, limit: int = 50, db: Session = Depends(get_db)
|
||||
|
|
|
|||
|
|
@ -127,6 +127,9 @@ class Settings(BaseSettings):
|
|||
biz_state_persist_every_cmds: int = 8
|
||||
# Cap raw_text loaded into Postgres from spool (0 = unlimited).
|
||||
biz_state_raw_max_bytes: int = 8 * 1024 * 1024
|
||||
# Manual log import: max upload / zip uncompressed bytes and text file count.
|
||||
biz_state_import_max_bytes: int = 256 * 1024 * 1024
|
||||
biz_state_import_max_files: int = 200
|
||||
# Dedicated biz_state worker process(es); general worker skips biz_state scheduler.
|
||||
biz_state_dedicated_workers: bool = True
|
||||
# Global ceiling for simultaneous running batches (across all workers).
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue