Initial Helpomatica source.
This commit is contained in:
@@ -0,0 +1,57 @@
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
|
||||
import log_preanalysis as lp
|
||||
|
||||
|
||||
def test_detect_job_and_date() -> None:
|
||||
query = "Разбери заказ 17876 за 2026-08-06: расхождение, Error64, промежутки, частота"
|
||||
assert lp.looks_like_log_job(query, ["line_u_codes.log.1"])
|
||||
assert lp.detect_job_id(query) == "17876"
|
||||
assert lp.detect_shift_date(query) == "2026-08-06"
|
||||
assert lp.detect_shift_date("смена 06.08.2026") == "2026-08-06"
|
||||
|
||||
|
||||
def test_stream_scan(tmp_path: Path) -> None:
|
||||
log = tmp_path / "line_u_codes.log.1"
|
||||
lines = [
|
||||
"2026-08-06 08:00:00 [Scanner] job=17876 code=010460406000000021ABC1",
|
||||
"2026-08-06 08:00:01 [PLC] job=17876 Error 12",
|
||||
"2026-08-06 08:05:00 [Scanner] job=17876 Error64 timeout",
|
||||
"2026-08-06 08:05:01 [PLC] Error65",
|
||||
"2026-08-06 10:00:00 [Scanner] last line",
|
||||
]
|
||||
log.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
percents: list[int] = []
|
||||
|
||||
def on_progress(info: dict) -> None:
|
||||
if info.get("percent") is not None:
|
||||
percents.append(int(info["percent"]))
|
||||
|
||||
report = lp.preanalyze_logs(
|
||||
[(log, "line_u_codes.log.1")],
|
||||
"заказ 17876 2026-08-06 Error64 промежутки частота",
|
||||
on_progress=on_progress,
|
||||
)
|
||||
assert report["job_id"] == "17876"
|
||||
item = report["files"][0]
|
||||
assert item["lines"] == 5
|
||||
assert item["error64"]
|
||||
assert item["error65"]
|
||||
assert item["gaps"]
|
||||
assert item["gaps"][0]["seconds"] >= 30
|
||||
md = report["markdown"]
|
||||
assert "PREANALYSIS" in md
|
||||
assert "Error64" in md
|
||||
text, encoding = lp.excerpt_lines(log, 2, 3)
|
||||
assert encoding == "utf-8"
|
||||
assert "2\t" in text and "Error 12" in text
|
||||
assert percents
|
||||
assert percents[-1] == 100
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_detect_job_and_date()
|
||||
with tempfile.TemporaryDirectory() as folder:
|
||||
test_stream_scan(Path(folder))
|
||||
print("ok")
|
||||
Reference in New Issue
Block a user