190 lines
7.2 KiB
Python
190 lines
7.2 KiB
Python
"""Непрерывный самоаудит omni-parse: собирает метрики, правит REPORT.md, копит аномалии.
|
||
|
||
Запуск: python tools/selfcheck.py [--json]
|
||
Ничего не меняет в базе, только читает и пишет REPORT.md + data/selfcheck.jsonl.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import py_compile
|
||
import subprocess
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT))
|
||
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
||
|
||
DB = ROOT / "data" / "omni.sqlite3"
|
||
REPORT = ROOT / "REPORT.md"
|
||
HIST = ROOT / "data" / "selfcheck.jsonl"
|
||
|
||
|
||
def q(con, sql, args=()):
|
||
try:
|
||
return con.execute(sql, args).fetchall()
|
||
except sqlite3.Error:
|
||
return []
|
||
|
||
|
||
def one(con, sql, args=(), default=0):
|
||
r = q(con, sql, args)
|
||
return r[0][0] if r and r[0][0] is not None else default
|
||
|
||
|
||
def run_tests():
|
||
"""Тесты — часть самоаудита: без них «работает» держится на памяти."""
|
||
try:
|
||
r = subprocess.run([sys.executable, "-m", "pytest", "tests", "-q"],
|
||
cwd=str(ROOT), capture_output=True, text=True, timeout=300)
|
||
except Exception as e:
|
||
return {"ok": False, "note": type(e).__name__}
|
||
tail = (r.stdout or "").strip().splitlines()
|
||
return {"ok": r.returncode == 0, "note": tail[-1][:120] if tail else ""}
|
||
|
||
|
||
def compile_all():
|
||
bad = []
|
||
for p in sorted((ROOT / "omni").glob("*.py")):
|
||
try:
|
||
py_compile.compile(str(p), doraise=True)
|
||
except py_compile.PyCompileError as e:
|
||
bad.append(f"{p.name}: {str(e).splitlines()[0][:120]}")
|
||
return bad
|
||
|
||
|
||
def collect():
|
||
st: dict = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds")}
|
||
st["compile_errors"] = compile_all()
|
||
st["tests"] = run_tests()
|
||
|
||
from omni import sources, taxonomy # noqa: E402
|
||
|
||
st["sources"] = len(sources.all_sources())
|
||
st["categories"] = len(sources.CATEGORIES)
|
||
|
||
if not DB.exists():
|
||
st["error"] = "нет базы"
|
||
return st, None
|
||
|
||
con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True, timeout=20)
|
||
try:
|
||
st["items"] = one(con, "SELECT count(*) FROM items")
|
||
st["priced"] = one(con, "SELECT count(*) FROM items WHERE price IS NOT NULL AND price > 0")
|
||
st["typed"] = one(con, "SELECT count(*) FROM items WHERE types IS NOT NULL AND types != ''")
|
||
st["runs"] = one(con, "SELECT count(*) FROM runs")
|
||
cols = [r[1] for r in q(con, "PRAGMA table_info(items)")]
|
||
st["has_types_col"] = "types" in cols
|
||
st["last_run"] = (q(con, "SELECT max(started_at) FROM runs") or [[None]])[0][0]
|
||
|
||
st["types"] = {k: v for k, v in q(
|
||
con, "SELECT types, count(*) FROM items WHERE types != '' GROUP BY types ORDER BY 2 DESC")}
|
||
flat = {}
|
||
for combo, n in st["types"].items():
|
||
for t in str(combo).split(","):
|
||
t = t.strip()
|
||
if t:
|
||
flat[t] = flat.get(t, 0) + n
|
||
st["type_flat"] = dict(sorted(flat.items(), key=lambda kv: -kv[1]))
|
||
st["type_labels"] = {k: taxonomy.LABELS.get(k, k) for k in st["type_flat"]}
|
||
|
||
st["by_source_top"] = [{"name": r[0], "n": r[1]} for r in q(
|
||
con, "SELECT source, count(*) FROM items GROUP BY 1 ORDER BY 2 DESC LIMIT 12")]
|
||
st["dupe_urls_sample"] = one(
|
||
con, "SELECT count(*) FROM (SELECT url FROM items WHERE url != '' GROUP BY url HAVING count(*) > 1)")
|
||
st["no_url"] = one(con, "SELECT count(*) FROM items WHERE url IS NULL OR url = ''")
|
||
st["empty_sources"] = st["sources"] - one(con, "SELECT count(DISTINCT source) FROM items")
|
||
st["abuse_stored"] = one(con, "SELECT count(*) FROM items WHERE category = 'abuse'") \
|
||
if "category" in cols else None
|
||
finally:
|
||
con.close()
|
||
|
||
problems = []
|
||
if st["compile_errors"]:
|
||
problems.append(f"compile: {len(st['compile_errors'])} файлов")
|
||
if not (st.get("tests") or {}).get("ok"):
|
||
problems.append(f"тесты: {(st.get('tests') or {}).get('note', 'fail')}")
|
||
if not st["has_types_col"]:
|
||
problems.append("нет колонки items.types")
|
||
if st["items"] and st["typed"] < st["items"] * 0.95:
|
||
problems.append(f"не типизировано {st['items'] - st['typed']} записей")
|
||
if st["dupe_urls_sample"]:
|
||
problems.append(f"дубли url: {st['dupe_urls_sample']}")
|
||
if st["sources"] and st["empty_sources"] > st["sources"] * 0.5:
|
||
problems.append(f"пустых источников {st['empty_sources']}/{st['sources']}")
|
||
st["problems"] = problems
|
||
return st, con
|
||
|
||
|
||
def render(st):
|
||
t = st.get("type_flat") or {}
|
||
labels = st.get("type_labels") or {}
|
||
types_line = " · ".join(f"{labels.get(k, k)} {v}" for k, v in list(t.items())[:8]) or "—"
|
||
top = st.get("by_source_top") or []
|
||
rows = "\n".join(f"| {r['name']} | {r['n']} |" for r in top) or "| — | 0 |"
|
||
problems = st.get("problems") or []
|
||
block = f"""<!-- STATS:START -->
|
||
_обновлено {st['ts']}_
|
||
|
||
| метрика | значение |
|
||
|---|---|
|
||
| источников | {st.get('sources', 0)} |
|
||
| категорий | {st.get('categories', 0)} |
|
||
| записей | {st.get('items', 0)} |
|
||
| с ценой | {st.get('priced', 0)} |
|
||
| типизировано | {st.get('typed', 0)} |
|
||
| прогонов | {st.get('runs', 0)} |
|
||
| пустых источников | {st.get('empty_sources', 0)} |
|
||
|
||
типы: {types_line}
|
||
|
||
| источник | записей |
|
||
|---|---|
|
||
{rows}
|
||
|
||
тесты: {(st.get('tests') or {}).get('note') or '-'}
|
||
самоаудит: {'OK' if not problems else '; '.join(problems)}
|
||
<!-- STATS:END -->"""
|
||
return block
|
||
|
||
|
||
def main():
|
||
st, _ = collect()
|
||
hist: list = []
|
||
HIST.parent.mkdir(parents=True, exist_ok=True)
|
||
if HIST.exists():
|
||
try:
|
||
hist = [json.loads(x) for x in HIST.read_text(encoding="utf-8").splitlines() if x.strip()]
|
||
except Exception:
|
||
hist = []
|
||
hist.append({k: st.get(k) for k in ("ts", "items", "priced", "typed", "runs", "sources", "problems")})
|
||
HIST.write_text("\n".join(json.dumps(x, ensure_ascii=False) for x in hist[-500:]) + "\n", encoding="utf-8")
|
||
|
||
block = render(st)
|
||
if REPORT.exists():
|
||
old = REPORT.read_text(encoding="utf-8")
|
||
if "<!-- STATS:START -->" in old and "<!-- STATS:END -->" in old:
|
||
head = old.split("<!-- STATS:START -->")[0]
|
||
tail = old.split("<!-- STATS:END -->")[1]
|
||
new = head + block + tail
|
||
else:
|
||
new = old.rstrip() + "\n\n" + block + "\n"
|
||
else:
|
||
new = "# omni-parse · автоотчёт\n\n" + block + "\n"
|
||
REPORT.write_text(new, encoding="utf-8")
|
||
|
||
if "--json" in sys.argv:
|
||
print(json.dumps(st, ensure_ascii=False, indent=1))
|
||
else:
|
||
print(f"items={st.get('items')} typed={st.get('typed')} sources={st.get('sources')} "
|
||
f"priced={st.get('priced')} runs={st.get('runs')} "
|
||
f"tests={(st.get('tests') or {}).get('ok')} problems={st.get('problems') or 'OK'}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main()) |