Files
Helpomatica/log_preanalysis.py
2026-08-14 11:06:58 +07:00

626 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Stream-scan InMark-style logs without loading the whole file into RAM."""
from __future__ import annotations
import gzip
import re
from collections import Counter
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Iterable
ProgressCb = Callable[[dict[str, Any]], None]
LOG_JOB_HINTS = re.compile(
r"job[_\s]?id|заказ|расхожден|error\s*6[45]|промежутк|частот|"
r"plc_errors|line_u|задани[еюя]|маркиров|отбраков|лог[аиов]?\b|"
r"histogram|ошибк|gap|код[аыов]?\b|смен[аыеу]",
re.IGNORECASE,
)
JOB_ID_RE = re.compile(
r"(?:job(?:[_\s-]?id)?|заказ(?:а|у|ом)?|задани[еюя]|jobid)\s*[:=#№]?\s*(\d{3,8})",
re.IGNORECASE,
)
STANDALONE_JOB_RE = re.compile(r"\b(\d{4,6})\b")
DATE_ISO_RE = re.compile(r"\b(20\d{2}-\d{2}-\d{2})\b")
DATE_RU_RE = re.compile(r"\b(\d{1,2})[./](\d{1,2})[./](20\d{2}|\d{2})\b")
TS_ISO_RE = re.compile(r"(20\d{2}-\d{2}-\d{2})[ T](\d{2}:\d{2}:\d{2})")
TS_RU_RE = re.compile(r"(\d{2}\.\d{2}\.20\d{2})[ T](\d{2}:\d{2}:\d{2})")
SOURCE_BRACKET_RE = re.compile(r"\[([A-Za-zА-Яа-яЁё0-9_./-]{2,40})\]")
SOURCE_FIELD_RE = re.compile(
r"\b(?:source|src|sender|from|channel|канал)\s*[=:]\s*([^\s,;\]=]+)",
re.IGNORECASE,
)
ERROR_RE = re.compile(r"\b(?:Error|error_?)\s*(\d{1,3})\b")
ERROR64_RE = re.compile(r"\b(?:Error|error_?)\s*64\b", re.IGNORECASE)
ERROR65_RE = re.compile(r"\b(?:Error|error_?)\s*65\b", re.IGNORECASE)
JOB_IN_LINE_RE = re.compile(
r"(?:job(?:[_\s-]?id)?|заказ|JobId)\s*[=:#]?\s*(\d{3,8})",
re.IGNORECASE,
)
CODE_RE = re.compile(
r"\b(?:01\d{14}21[A-Za-z0-9]{1,24}|[A-F0-9]{24,44}|\d{20,32})\b"
)
LOG_NAME_RE = re.compile(r"\.log(\.\d+)?(\.gz)?$", re.IGNORECASE)
PREFERRED_LOG_NAMES = (
"line_u_codes.log",
"plc_errors.log",
"line_u.log",
"line.log",
)
GAP_SECONDS = 30
MAX_UNIQUE_CODES = 20_000
MAX_ERROR64_HITS = 80
MAX_GAPS = 25
MAX_SAMPLE_LINES = 12
MAX_PREANALYSIS_CHARS = 70_000
READ_CHUNK = 256 * 1024
def looks_like_log_job(query: str, log_file_names: Iterable[str] | None = None) -> bool:
text = str(query or "")
if LOG_JOB_HINTS.search(text):
return True
names = [str(name or "") for name in (log_file_names or [])]
if names and (JOB_ID_RE.search(text) or DATE_ISO_RE.search(text) or DATE_RU_RE.search(text)):
return True
return False
def is_log_filename(name: str) -> bool:
lower = str(name or "").lower()
if LOG_NAME_RE.search(lower):
return True
return ".log." in lower or lower.endswith(".log")
def detect_job_id(query: str) -> str | None:
text = str(query or "")
match = JOB_ID_RE.search(text)
if match:
return match.group(1)
hinted = LOG_JOB_HINTS.search(text)
if hinted:
standalone = STANDALONE_JOB_RE.search(text)
if standalone:
return standalone.group(1)
return None
def detect_shift_date(query: str) -> str | None:
text = str(query or "")
iso = DATE_ISO_RE.search(text)
if iso:
return iso.group(1)
ru = DATE_RU_RE.search(text)
if not ru:
return None
day, month, year = ru.group(1), ru.group(2), ru.group(3)
if len(year) == 2:
year = f"20{year}"
try:
return datetime(int(year), int(month), int(day)).strftime("%Y-%m-%d")
except ValueError:
return None
def detect_log_encoding(path: Path) -> str:
try:
raw = path.open("rb").read(65536)
except OSError:
return "utf-8"
if raw.startswith(b"\xef\xbb\xbf"):
return "utf-8-sig"
for encoding in ("utf-8", "cp1251"):
try:
raw.decode(encoding)
return encoding
except UnicodeDecodeError:
continue
return "cp1251"
def _parse_timestamp(line: str) -> datetime | None:
match = TS_ISO_RE.search(line)
if match:
try:
return datetime.strptime(
f"{match.group(1)} {match.group(2)}", "%Y-%m-%d %H:%M:%S"
)
except ValueError:
return None
match = TS_RU_RE.search(line)
if not match:
return None
try:
return datetime.strptime(
f"{match.group(1)} {match.group(2)}", "%d.%m.%Y %H:%M:%S"
)
except ValueError:
return None
def _source_of(line: str, fallback: str) -> str:
field = SOURCE_FIELD_RE.search(line)
if field:
return field.group(1)[:40]
bracket = SOURCE_BRACKET_RE.search(line)
if bracket:
token = bracket.group(1)
if not re.fullmatch(r"\d{1,2}:\d{2}:\d{2}.*", token):
return token[:40]
return fallback
def _open_log_binary(path: Path):
name = path.name.lower()
handle = path.open("rb")
if name.endswith(".gz"):
return gzip.GzipFile(fileobj=handle, mode="rb"), handle
return handle, handle
def iter_log_lines(path: Path, encoding: str):
"""Yield (line_no, text, bytes_read, total_bytes) without loading the file."""
size = 0
try:
size = path.stat().st_size
except OSError:
size = 0
stream, raw = _open_log_binary(path)
try:
leftover = b""
line_no = 0
bytes_read = 0
while True:
chunk = stream.read(READ_CHUNK)
if not chunk:
break
bytes_read += len(chunk)
data = leftover + chunk
parts = data.split(b"\n")
leftover = parts.pop()
for part in parts:
line_no += 1
text = part.rstrip(b"\r").decode(encoding, errors="replace")
yield line_no, text, min(bytes_read, size) if size else bytes_read, size
if leftover:
line_no += 1
text = leftover.rstrip(b"\r").decode(encoding, errors="replace")
yield line_no, text, size or bytes_read, size
finally:
try:
stream.close()
except Exception: # noqa: BLE001
pass
if raw is not stream:
try:
raw.close()
except Exception: # noqa: BLE001
pass
def rank_log_files(entries: list[dict]) -> list[dict]:
def score(entry: dict) -> tuple:
name = str(entry.get("name") or "").lower()
preferred = 0
for index, prefix in enumerate(PREFERRED_LOG_NAMES):
if name.startswith(prefix) or prefix in name:
preferred = 10 - index
break
size = int(entry.get("size") or 0)
return (preferred, 1 if is_log_filename(name) else 0, min(size, 10**12))
logs = [entry for entry in entries if is_log_filename(str(entry.get("name") or ""))]
logs.sort(key=score, reverse=True)
return logs[:6]
def _emit_progress(
on_progress: ProgressCb | None,
*,
phase: str,
message: str,
percent: int | None,
file: str = "",
) -> None:
if not on_progress:
return
payload: dict[str, Any] = {
"phase": phase,
"message": message,
"percent": percent,
}
if file:
payload["file"] = file
on_progress(payload)
def _scan_one_file(
path: Path,
display_name: str,
job_id: str | None,
shift_date: str | None,
*,
on_progress: ProgressCb | None = None,
overall_offset: int = 0,
overall_total: int = 1,
cancel: Callable[[], bool] | None = None,
) -> dict[str, Any]:
encoding = detect_log_encoding(path)
sources: Counter[str] = Counter()
errors: Counter[str] = Counter()
hours: Counter[str] = Counter()
unique_codes: set[str] = set()
unique_capped = False
code_events = 0
error64: list[dict[str, Any]] = []
error65: list[dict[str, Any]] = []
gaps: list[dict[str, Any]] = []
samples: list[dict[str, Any]] = []
first_ts: datetime | None = None
last_ts: datetime | None = None
prev_ts: datetime | None = None
prev_line = 0
line_count = 0
job_lines = 0
shift_lines = 0
last_pct = -1
def maybe_sample(line_no: int, line: str, reason: str) -> None:
if len(samples) >= MAX_SAMPLE_LINES:
return
samples.append(
{
"file": display_name,
"line": line_no,
"text": line[:240],
"reason": reason,
}
)
for line_no, line, bytes_read, total in iter_log_lines(path, encoding):
if cancel and cancel():
break
line_count = line_no
if overall_total:
local = (bytes_read / total) if total else 1.0
overall = overall_offset + local * (total or 0)
pct = min(99, int(100 * overall / overall_total))
if pct >= last_pct + 2:
last_pct = pct
_emit_progress(
on_progress,
phase="scan",
message=f"Читаю лог… {pct}%",
percent=pct,
file=display_name,
)
ts = _parse_timestamp(line)
if ts:
if first_ts is None:
first_ts = ts
maybe_sample(line_no, line, "first")
last_ts = ts
hours[ts.strftime("%Y-%m-%d %H:00")] += 1
if prev_ts is not None:
delta = (ts - prev_ts).total_seconds()
if delta >= GAP_SECONDS:
gaps.append(
{
"file": display_name,
"from_line": prev_line,
"to_line": line_no,
"from_ts": prev_ts.isoformat(sep=" ", timespec="seconds"),
"to_ts": ts.isoformat(sep=" ", timespec="seconds"),
"seconds": int(delta),
}
)
prev_ts = ts
prev_line = line_no
in_shift = True
if shift_date and ts:
in_shift = ts.strftime("%Y-%m-%d") == shift_date
elif shift_date and shift_date in line:
in_shift = True
elif shift_date:
in_shift = False
if in_shift:
shift_lines += 1
job_hit = JOB_IN_LINE_RE.search(line)
if job_id and (job_id in line or (job_hit and job_hit.group(1) == job_id)):
job_lines += 1
elif job_hit:
job_lines += 0
sources[_source_of(line, display_name)] += 1
for err in ERROR_RE.findall(line):
if in_shift or not shift_date:
errors[err] += 1
if ERROR64_RE.search(line) and len(error64) < MAX_ERROR64_HITS:
error64.append(
{
"file": display_name,
"line": line_no,
"ts": ts.isoformat(sep=" ", timespec="seconds") if ts else "",
"text": line[:220],
}
)
maybe_sample(line_no, line, "Error64")
if ERROR65_RE.search(line) and len(error65) < MAX_ERROR64_HITS:
error65.append(
{
"file": display_name,
"line": line_no,
"ts": ts.isoformat(sep=" ", timespec="seconds") if ts else "",
"text": line[:220],
}
)
maybe_sample(line_no, line, "Error65")
for code in CODE_RE.findall(line):
code_events += 1
if not unique_capped:
unique_codes.add(code)
if len(unique_codes) >= MAX_UNIQUE_CODES:
unique_capped = True
if last_ts and line_count:
maybe_sample(line_count, "", "last")
if samples and samples[-1]["reason"] == "last" and not samples[-1]["text"]:
samples.pop()
gaps.sort(key=lambda item: int(item.get("seconds") or 0), reverse=True)
gaps = gaps[:MAX_GAPS]
return {
"file": display_name,
"path": str(path),
"encoding": encoding,
"lines": line_count,
"job_lines": job_lines,
"shift_lines": shift_lines,
"first_ts": first_ts.isoformat(sep=" ", timespec="seconds") if first_ts else "",
"last_ts": last_ts.isoformat(sep=" ", timespec="seconds") if last_ts else "",
"sources": sources.most_common(30),
"errors": errors.most_common(25),
"hours": hours.most_common(24),
"unique_codes": len(unique_codes),
"unique_capped": unique_capped,
"code_events": code_events,
"error64": error64,
"error65": error65,
"gaps": gaps,
"samples": samples,
}
def _format_duration(seconds: int) -> str:
value = max(0, int(seconds))
if value < 60:
return f"{value} с"
minutes, rest = divmod(value, 60)
if minutes < 60:
return f"{minutes} мин {rest} с" if rest else f"{minutes} мин"
hours, minutes = divmod(minutes, 60)
return f"{hours} ч {minutes} мин"
def render_preanalysis_markdown(report: dict[str, Any]) -> str:
job_id = report.get("job_id") or "не найден в вопросе"
shift = report.get("shift_date") or "не указана"
lines: list[str] = [
"# PREANALYSIS",
"",
"Сервер уже просканировал логи **одним проходом**. "
"Используй этот файл первым. Не читай логи целиком заново, "
"если цифры выглядят правдоподобно. Для проверки бери фрагмент "
"по номеру строки (`:15231:line_u_codes.log.1` или "
"`::log:line_u_codes.log.1:15231::`).",
"",
f"- Job id: `{job_id}`",
f"- Дата смены (из вопроса): `{shift}`",
f"- Файлов просканировано: {len(report.get('files') or [])}",
"",
]
for item in report.get("files") or []:
name = item.get("file") or "log"
lines.extend(
[
f"## {name}",
"",
f"- Кодировка: `{item.get('encoding')}`",
f"- Строк: {item.get('lines')}",
f"- Строк с job id: {item.get('job_lines')}",
f"- Строк за смену: {item.get('shift_lines')}",
f"- Интервал: {item.get('first_ts') or '—'}{item.get('last_ts') or '—'}",
f"- Событий с кодами: {item.get('code_events')}",
f"- Уникальных кодов: {item.get('unique_codes')}"
+ (" (учёт ограничен) " if item.get("unique_capped") else ""),
"",
"### Источники",
"",
]
)
sources = item.get("sources") or []
if sources:
for src, count in sources:
lines.append(f"- `{src}`: {count}")
else:
lines.append("- (нет)")
lines.extend(["", "### Ошибки за смену (гистограмма)", ""])
errors = item.get("errors") or []
if errors:
for code, count in errors:
lines.append(f"- Error {code}: {count}")
else:
lines.append("- (нет совпадений Error N)")
lines.extend(["", "### Частота по часам (топ)", ""])
hours = item.get("hours") or []
if hours:
for hour, count in hours[:16]:
lines.append(f"- {hour}: {count}")
else:
lines.append("- (нет меток времени)")
lines.extend(["", "### Длинные промежутки (≥ 30 с)", ""])
gaps = item.get("gaps") or []
if gaps:
for gap in gaps:
lines.append(
f"- {_format_duration(int(gap.get('seconds') or 0))}: "
f"{gap.get('from_ts')}{gap.get('to_ts')} "
f"(`:{gap.get('from_line')}:{name}` … `:{gap.get('to_line')}:{name}`)"
)
else:
lines.append("- нет промежутков ≥ 30 с")
lines.extend(["", "### Error64", ""])
hits64 = item.get("error64") or []
if hits64:
for hit in hits64[:40]:
ts = hit.get("ts") or "—"
lines.append(
f"- {ts} `:{hit.get('line')}:{name}` — {hit.get('text')}"
)
else:
lines.append("- нет")
lines.extend(["", "### Error65", ""])
hits65 = item.get("error65") or []
if hits65:
for hit in hits65[:40]:
ts = hit.get("ts") or "—"
lines.append(
f"- {ts} `:{hit.get('line')}:{name}` — {hit.get('text')}"
)
else:
lines.append("- нет")
lines.extend(["", "### Примеры строк", ""])
for sample in item.get("samples") or []:
text = str(sample.get("text") or "").strip()
if not text:
continue
lines.append(
f"- `:{sample.get('line')}:{name}` ({sample.get('reason')}): `{text}`"
)
lines.append("")
text = "\n".join(lines).strip() + "\n"
if len(text) > MAX_PREANALYSIS_CHARS:
text = text[:MAX_PREANALYSIS_CHARS].rstrip() + "\n\n\n"
return text
def preanalyze_logs(
files: list[tuple[Path, str]],
query: str,
*,
on_progress: ProgressCb | None = None,
cancel: Callable[[], bool] | None = None,
) -> dict[str, Any]:
job_id = detect_job_id(query)
shift_date = detect_shift_date(query)
existing = [(path, name) for path, name in files if path.exists() and path.is_file()]
total_bytes = 0
for path, _name in existing:
try:
total_bytes += path.stat().st_size
except OSError:
pass
total_bytes = max(total_bytes, 1)
offset = 0
scanned: list[dict[str, Any]] = []
_emit_progress(
on_progress,
phase="scan",
message="Читаю лог… 0%",
percent=0,
)
for path, name in existing:
size = 0
try:
size = path.stat().st_size
except OSError:
size = 0
item = _scan_one_file(
path,
name,
job_id,
shift_date,
on_progress=on_progress,
overall_offset=offset,
overall_total=total_bytes,
cancel=cancel,
)
scanned.append(item)
offset += size
if cancel and cancel():
break
_emit_progress(
on_progress,
phase="scan",
message="Читаю лог… 100%",
percent=100,
)
report = {
"job_id": job_id or "",
"shift_date": shift_date or "",
"files": scanned,
}
report["markdown"] = render_preanalysis_markdown(report)
report["summary"] = _short_summary(report)
return report
def _short_summary(report: dict[str, Any]) -> str:
bits = []
if report.get("job_id"):
bits.append(f"job {report['job_id']}")
if report.get("shift_date"):
bits.append(report["shift_date"])
for item in report.get("files") or []:
bits.append(
f"{item.get('file')}: {item.get('lines')} строк, "
f"Error64×{len(item.get('error64') or [])}, "
f"промежутков {len(item.get('gaps') or [])}"
)
return "; ".join(bits) if bits else "логи просканированы"
def excerpt_lines(
path: Path,
start: int,
end: int,
encoding: str | None = None,
) -> tuple[str, str]:
"""Return (text, encoding) for 1-indexed inclusive line range."""
start = max(1, int(start))
end = max(start, int(end))
encoding = encoding or detect_log_encoding(path)
collected: list[str] = []
for line_no, text, _read, _total in iter_log_lines(path, encoding):
if line_no < start:
continue
if line_no > end:
break
collected.append(f"{line_no}\t{text}")
return "\n".join(collected) + ("\n" if collected else ""), encoding
def write_preanalysis_md(workspace: Path, markdown: str) -> Path:
target = Path(workspace) / "PREANALYSIS.md"
target.write_text(markdown, encoding="utf-8")
listing = Path(workspace) / "PROJECT_FILES.md"
if listing.exists():
extra = "\n- ./PREANALYSIS.md — server-side log scan; read this first\n"
text = listing.read_text(encoding="utf-8")
if "PREANALYSIS.md" not in text:
listing.write_text(text.rstrip() + extra, encoding="utf-8")
return target