178 lines
5.4 KiB
Python
178 lines
5.4 KiB
Python
"""监控检查纯函数核(spec §20.15.2 四层证据链的结果层+过程层解析件).
|
|
|
|
约定: 交易日历=list['YYYY-MM-DD'] 升序; 期望值='YYYYMMDD'(与存盘 stem 对齐);
|
|
now 的 ready 语义=「数据当日就绪时刻」——now 早于 ready 则该日尚不该有数据。
|
|
"""
|
|
import datetime as _dt
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sqlite3
|
|
|
|
|
|
# ---------- 交易日历(dbbardata 沪深300 日线当锚; 仓库无独立日历表) ----------
|
|
|
|
def trading_days(db_path, today, lookback=50):
|
|
"""近 lookback 日历日的交易日序列; db 不可达返 [](调用方降级周一~五启发式)."""
|
|
cutoff = (today - _dt.timedelta(days=lookback)).isoformat()
|
|
try:
|
|
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=10)
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT DISTINCT datetime FROM dbbardata "
|
|
"WHERE symbol='000300' AND exchange='SSE' AND interval='d' "
|
|
"AND datetime>=? ORDER BY datetime", (cutoff,)).fetchall()
|
|
finally:
|
|
conn.close()
|
|
return [r[0] for r in rows]
|
|
except sqlite3.Error:
|
|
return []
|
|
|
|
|
|
def _hhmm(ready):
|
|
h, m = (int(x) for x in ready.split(":"))
|
|
return h * 60 + m
|
|
|
|
|
|
def expected_daily(days, now, ready="20:30", lag=0):
|
|
if not days:
|
|
return None
|
|
now_min = now.hour * 60 + now.minute
|
|
today = now.date().isoformat()
|
|
ok = [d for d in days if d < today
|
|
or (d == today and now_min >= _hhmm(ready))]
|
|
if len(ok) <= lag:
|
|
return None
|
|
return ok[-1 - lag].replace("-", "")
|
|
|
|
|
|
def expected_weekly(days, now, ready="06:00"):
|
|
"""最后周五锚 F: 当周六 ready 后 F 可验(ak-weekly 周六 03:00 产出)."""
|
|
if not days:
|
|
return None
|
|
deadline = now - _dt.timedelta(days=1)
|
|
floor = deadline.date() if now.hour * 60 + now.minute >= _hhmm(ready) \
|
|
else deadline.date() - _dt.timedelta(days=1)
|
|
fridays = [d for d in days
|
|
if _dt.date.fromisoformat(d).weekday() == 4
|
|
and d <= floor.isoformat()]
|
|
return fridays[-1].replace("-", "") if fridays else None
|
|
|
|
|
|
# ---------- 目录/文件 ----------
|
|
|
|
def latest_stem(directory, pattern):
|
|
if not os.path.isdir(directory):
|
|
return None
|
|
rx = re.compile(pattern)
|
|
best = None
|
|
for fn in os.listdir(directory):
|
|
stem = fn.rsplit(".", 1)[0] if "." in fn else fn
|
|
m = rx.match(stem)
|
|
if m and (best is None or stem > best[0]):
|
|
best = (stem, os.path.join(directory, fn))
|
|
return best
|
|
|
|
|
|
def parquet_rows(path):
|
|
try:
|
|
import pyarrow.parquet as pq
|
|
return pq.ParquetFile(path).metadata.num_rows
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def dir_latest_mtime(directory):
|
|
latest = 0.0
|
|
for root, _dirs, files in os.walk(directory):
|
|
for fn in files:
|
|
try:
|
|
latest = max(latest, os.path.getmtime(os.path.join(root, fn)))
|
|
except OSError:
|
|
pass
|
|
return latest
|
|
|
|
|
|
def check_disk(path):
|
|
target = path if os.path.exists(path) else os.path.dirname(os.path.abspath(path))
|
|
usage = shutil.disk_usage(target)
|
|
free_gb = usage.free / 1024 ** 3
|
|
if free_gb < 8:
|
|
status = "red"
|
|
elif free_gb < 20:
|
|
status = "yellow"
|
|
else:
|
|
status = "green"
|
|
return {"free_gb": round(free_gb, 1), "status": status}
|
|
|
|
|
|
# ---------- 过程层: 统计行+失败签名分诊(spec §20.15.2 ②) ----------
|
|
|
|
_STAT_RE = re.compile(r"\[(\w+)\] 完成.*?统计: (\{[^\n]*\})")
|
|
|
|
SIGNATURES = [
|
|
("断路器触发", "breaker", "red"),
|
|
("RemoteDisconnected", "disconnect", "red"),
|
|
("Expecting value", "source_changed", "red"),
|
|
("418", "anti_crawl", "red"),
|
|
("'NoneType' object is not subscriptable", "source_empty", "yellow"),
|
|
("KeyError", "field_changed", "red"),
|
|
]
|
|
|
|
|
|
def parse_stat_lines(text):
|
|
import json
|
|
out = {}
|
|
for m in _STAT_RE.finditer(text):
|
|
try:
|
|
out[m.group(1)] = json.loads(m.group(2))
|
|
except ValueError:
|
|
continue
|
|
return out
|
|
|
|
|
|
def scan_signatures(text):
|
|
seen, out = set(), []
|
|
for needle, sig, severity in SIGNATURES:
|
|
if needle in text and sig not in seen:
|
|
seen.add(sig)
|
|
out.append({"sig": sig, "severity": severity})
|
|
return out
|
|
|
|
|
|
# ---------- 调度层: schtasks /query /fo LIST /v 解析(双语字段名) ----------
|
|
|
|
_FIELD_PATTERNS = {
|
|
"task_name": r"^(?:TaskName|任务名)\s*:\s*(.*)$",
|
|
"next_run": r"^(?:Next Run Time|下次运行时间)\s*:\s*(.*)$",
|
|
"status": r"^(?:Status|状态)\s*:\s*(.*)$",
|
|
"last_run": r"^(?:Last Run Time|上次运行时间)\s*:\s*(.*)$",
|
|
"last_result": r"^(?:Last Result|上次结果)\s*:\s*(.*)$",
|
|
}
|
|
_COMPILED = {k: re.compile(p) for k, p in _FIELD_PATTERNS.items()}
|
|
|
|
|
|
def parse_schtasks_list(text):
|
|
out = {}
|
|
cur = None
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
matched = False
|
|
for key, rx in _COMPILED.items():
|
|
m = rx.match(stripped)
|
|
if not m:
|
|
continue
|
|
val = m.group(1).strip()
|
|
if key == "task_name":
|
|
cur = val.lstrip("\\")
|
|
out.setdefault(cur, {})
|
|
elif cur is not None:
|
|
out[cur][key] = val
|
|
matched = True
|
|
break
|
|
if not matched and cur is None:
|
|
continue # HostName 等无关行
|
|
return out
|