butubb
f6ddc46d62
问题：cookie 过期导致任务失败，但没有任何通知。查下来不是代码问题 —— notify_enabled 在两个任务上都是 False，而它默认就是关的，事件（run_failed） 也确实生成了，只卡在最后一道闸门。 但那个默认值是错的。代码里的理由是「一条任务列表都推到一个群会很快变吵，所以默认静默」， 这个理由对新作品成立（可能每轮都有），对失败不成立：一次登录态失效意味着这个任务事实上 已经死了，而你不会知道，直到某天发现数据停在几周前。最该被告知的就是这种情况。 现在拆开： - notify_enabled —— 推送新作品，可能每轮都有，默认关 - notify_failures —— 推送异常（登录失效/运行失败/没抓到数据），默认开 事件按开关过滤（build_run_message）：只勾了「新作品」的任务不该因为一次失败被推消息， 反之亦然，否则拆开开关就没有意义。已有任务由 _ensure_columns 补上 notify_failures=1， 所以会自动开始收到异常推送。 列名 notify_enabled 是历史遗留（它早先是唯一的通知开关），语义已收窄为「新作品」， 用注释写明，不做列重命名 —— 那需要单独的迁移，不值为一个内部工具做。
825 lines
30 KiB
Python
825 lines
30 KiB
Python
# -*- coding: utf-8 -*-
|
||
# Copyright (c) 2025 [email protected]
|
||
#
|
||
# This file is part of MediaCrawler project.
|
||
# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/api/monitor/service.py
|
||
# GitHub: https://github.com/NanmiCoder
|
||
# Non-commercial learning license 1.1
|
||
#
|
||
# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
|
||
# 1. 不得用于任何商业用途。
|
||
# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
|
||
# 3. 不得进行大规模爬取或对平台造成运营干扰。
|
||
# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
|
||
# 5. 不得用于任何非法或不当的用途。
|
||
#
|
||
# 详细许可条款请参阅项目根目录下的LICENSE文件。
|
||
# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。
|
||
|
||
"""Task CRUD and dashboard queries for the monitoring layer."""
|
||
|
||
import asyncio
|
||
import re
|
||
from typing import Any, Dict, List, Optional
|
||
from urllib.parse import parse_qs, urlparse
|
||
|
||
from sqlalchemy import delete, func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from tools.time_util import get_current_timestamp
|
||
|
||
from . import app_settings, covers, platforms, schedule
|
||
from .db import get_session
|
||
from .platforms import PLATFORM_XHS
|
||
from .models import (
|
||
MODE_CREATOR,
|
||
MODE_NOTE,
|
||
MonitorComment,
|
||
MonitorEvent,
|
||
MonitorNote,
|
||
MonitorNoteMetric,
|
||
MonitorRun,
|
||
MonitorTarget,
|
||
MonitorTask,
|
||
RUN_SUCCESS,
|
||
RUN_PARTIAL,
|
||
)
|
||
from .runner import execute_task
|
||
|
||
# Keep strong references to in-flight manual runs; asyncio only holds weak ones,
|
||
# so without this a run can be garbage collected mid-flight.
|
||
_background_runs: set[asyncio.Task] = set()
|
||
|
||
MIN_INTERVAL_MINUTES = 30
|
||
MAX_INTERVAL_MINUTES = 7 * 24 * 60
|
||
|
||
# Changing any of these invalidates the pending run slot.
|
||
SCHEDULE_FIELDS = {
|
||
"interval_minutes",
|
||
"schedule_mode",
|
||
"schedule_hours",
|
||
"schedule_days",
|
||
"schedule_minute",
|
||
}
|
||
|
||
|
||
def _require_clock_fields(mode: str, hours: list, days: list) -> None:
|
||
"""A clock schedule with no clock time can never fire.
|
||
|
||
The create schema already rejects that shape, but the update path merges
|
||
partial fields and therefore has no schema-level view of the result -- so the
|
||
check lives here, where both paths meet.
|
||
"""
|
||
if mode in schedule.CLOCK_MODES and not hours:
|
||
raise ValueError("按钟点调度至少要选一个时间")
|
||
if mode == schedule.MODE_WEEKLY and not days:
|
||
raise ValueError("按周调度至少要选一个星期")
|
||
|
||
_CREATOR_URL_RE = re.compile(r"xiaohongshu\.com/user/profile/([A-Za-z0-9_-]+)")
|
||
_NOTE_URL_RE = re.compile(r"xiaohongshu\.com/(?:explore|discovery/item)/([A-Za-z0-9_-]+)")
|
||
# XHS user ids and note ids are 24-char hex; allow a slightly wider range so a
|
||
# format change degrades into "still accepted" rather than "rejected".
|
||
_BARE_ID_RE = re.compile(r"^[A-Za-z0-9_-]{8,64}$")
|
||
|
||
|
||
class TargetParseError(ValueError):
|
||
"""Raised when a pasted monitoring target cannot be understood."""
|
||
|
||
|
||
def parse_target_input(
|
||
value: str, mode: str, platform: str = PLATFORM_XHS
|
||
) -> Dict[str, str]:
|
||
"""Parse a pasted creator/note value into a stable id plus a refreshable token.
|
||
|
||
Accepts either a full URL (with or without ``xsec_token``) or a bare id.
|
||
Storing the id separately from the token is what keeps a long-running task
|
||
alive: tokens expire, ids do not.
|
||
|
||
URL shapes are platform-specific. Only Xiaohongshu is wired, so anything else
|
||
is rejected here as well as at task creation -- parsing a Douyin link as if it
|
||
were a Xiaohongshu one would be worse than refusing it.
|
||
"""
|
||
if platform != PLATFORM_XHS:
|
||
raise TargetParseError(f"暂不支持解析该平台（{platform}）的目标链接")
|
||
|
||
raw = (value or "").strip()
|
||
if not raw:
|
||
raise TargetParseError("Empty target")
|
||
|
||
external_id = ""
|
||
if raw.startswith("http") or "/" in raw:
|
||
# xhslink.com and other short links are not resolvable without a network
|
||
# round-trip, so only the direct profile/explore forms are supported.
|
||
match = _CREATOR_URL_RE.search(raw) if mode == MODE_CREATOR else _NOTE_URL_RE.search(raw)
|
||
if not match:
|
||
expected = "博主主页" if mode == MODE_CREATOR else "笔记"
|
||
raise TargetParseError(f"无法从链接中解析出{expected} ID：{raw}")
|
||
external_id = match.group(1)
|
||
elif _BARE_ID_RE.match(raw):
|
||
external_id = raw
|
||
else:
|
||
raise TargetParseError(f"无法识别的目标：{raw}")
|
||
|
||
params = parse_qs(urlparse(raw).query) if raw.startswith("http") else {}
|
||
return {
|
||
"external_id": external_id,
|
||
"xsec_token": (params.get("xsec_token") or [""])[0],
|
||
"xsec_source": (params.get("xsec_source") or [""])[0],
|
||
"raw_value": raw,
|
||
}
|
||
|
||
|
||
async def platform_task_ids(session: AsyncSession, platform: str) -> List[int]:
|
||
"""Ids of the tasks belonging to a platform.
|
||
|
||
Note/comment/event tables carry no platform column -- they hang off a task --
|
||
so scoping a query to a platform means scoping it to that task set.
|
||
"""
|
||
return list(
|
||
await session.scalars(select(MonitorTask.id).where(MonitorTask.platform == platform))
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Task CRUD
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def create_task(session: AsyncSession, payload: Dict[str, Any]) -> MonitorTask:
|
||
mode = payload["mode"]
|
||
if mode not in (MODE_CREATOR, MODE_NOTE):
|
||
raise ValueError(f"Unsupported mode: {mode}")
|
||
|
||
# Rejects unknown platforms and, more importantly, platforms whose crawler
|
||
# exists upstream but whose monitoring is not wired up -- accepting those
|
||
# would create a task that can never produce data.
|
||
platform = payload.get("platform") or PLATFORM_XHS
|
||
platforms.ensure_runnable(platform)
|
||
|
||
now = get_current_timestamp()
|
||
|
||
# Fall back to the configured defaults for anything the caller left out, so
|
||
# the Settings page actually governs new tasks.
|
||
defaults = await app_settings.defaults(session, platform)
|
||
interval_minutes = payload.get("interval_minutes") or defaults["interval_minutes"]
|
||
|
||
schedule_mode = payload.get("schedule_mode") or schedule.MODE_INTERVAL
|
||
schedule_hours = list(payload.get("schedule_hours") or [])
|
||
schedule_days = list(payload.get("schedule_days") or [])
|
||
schedule_minute = int(payload.get("schedule_minute") or 0)
|
||
_require_clock_fields(schedule_mode, schedule_hours, schedule_days)
|
||
|
||
task = MonitorTask(
|
||
name=payload["name"],
|
||
platform=platform,
|
||
mode=mode,
|
||
enabled=payload.get("enabled", True),
|
||
interval_minutes=interval_minutes,
|
||
schedule_mode=schedule_mode,
|
||
schedule_hours=schedule.format_hours(schedule_hours),
|
||
schedule_days=schedule.format_days(schedule_days),
|
||
schedule_minute=schedule_minute,
|
||
max_notes_count=payload.get("max_notes_count") or defaults["max_notes_count"],
|
||
enable_comments=payload.get("enable_comments", True),
|
||
max_comments_count=payload.get("max_comments_count") or defaults["max_comments_count"],
|
||
run_timeout_seconds=payload.get("run_timeout_seconds", 3600),
|
||
notify_enabled=payload.get("notify_enabled", False),
|
||
notify_failures=payload.get("notify_failures", True),
|
||
next_run_at=schedule.next_occurrence(
|
||
mode=schedule_mode,
|
||
interval_minutes=interval_minutes,
|
||
hours=schedule_hours,
|
||
days=schedule_days,
|
||
minute=schedule_minute,
|
||
after_ms=now,
|
||
),
|
||
last_status="idle",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
session.add(task)
|
||
await session.flush()
|
||
|
||
seen: set[str] = set()
|
||
for value in payload.get("targets", []):
|
||
parsed = parse_target_input(value, mode, platform)
|
||
if parsed["external_id"] in seen:
|
||
continue
|
||
seen.add(parsed["external_id"])
|
||
session.add(
|
||
MonitorTarget(
|
||
task_id=task.id,
|
||
kind=mode,
|
||
external_id=parsed["external_id"],
|
||
xsec_token=parsed["xsec_token"],
|
||
xsec_source=parsed["xsec_source"],
|
||
raw_value=parsed["raw_value"],
|
||
label=parsed["external_id"],
|
||
enabled=True,
|
||
created_at=now,
|
||
)
|
||
)
|
||
|
||
await session.flush()
|
||
return task
|
||
|
||
|
||
async def update_task(session: AsyncSession, task_id: int, payload: Dict[str, Any]) -> MonitorTask:
|
||
task = await session.get(MonitorTask, task_id)
|
||
if task is None:
|
||
raise ValueError(f"Task {task_id} not found")
|
||
|
||
was_enabled = task.enabled
|
||
|
||
for field in (
|
||
"name",
|
||
"enabled",
|
||
"interval_minutes",
|
||
"schedule_mode",
|
||
"schedule_minute",
|
||
"max_notes_count",
|
||
"enable_comments",
|
||
"max_comments_count",
|
||
"run_timeout_seconds",
|
||
"notify_enabled",
|
||
"notify_failures",
|
||
):
|
||
if field in payload and payload[field] is not None:
|
||
setattr(task, field, payload[field])
|
||
|
||
# The clock lists are stored as comma-separated text, so they cannot go
|
||
# through the generic loop above.
|
||
if payload.get("schedule_hours") is not None:
|
||
task.schedule_hours = schedule.format_hours(payload["schedule_hours"])
|
||
if payload.get("schedule_days") is not None:
|
||
task.schedule_days = schedule.format_days(payload["schedule_days"])
|
||
|
||
# Replacing targets resets the baseline implicitly: a note set that now
|
||
# includes new ids will simply report them as new on the next run.
|
||
if payload.get("targets") is not None:
|
||
await session.execute(delete(MonitorTarget).where(MonitorTarget.task_id == task_id))
|
||
now = get_current_timestamp()
|
||
seen: set[str] = set()
|
||
for value in payload["targets"]:
|
||
parsed = parse_target_input(value, task.mode)
|
||
if parsed["external_id"] in seen:
|
||
continue
|
||
seen.add(parsed["external_id"])
|
||
session.add(
|
||
MonitorTarget(
|
||
task_id=task_id,
|
||
kind=task.mode,
|
||
external_id=parsed["external_id"],
|
||
xsec_token=parsed["xsec_token"],
|
||
xsec_source=parsed["xsec_source"],
|
||
raw_value=parsed["raw_value"],
|
||
label=parsed["external_id"],
|
||
enabled=True,
|
||
created_at=now,
|
||
)
|
||
)
|
||
|
||
# Any change to when the task runs invalidates the pending slot, so recompute
|
||
# it from the merged state rather than working out which field moved.
|
||
# Re-enabling counts as a change too: otherwise a task switched off for a
|
||
# month comes back holding a next_run_at a month in the past and fires the
|
||
# instant it is saved.
|
||
if (SCHEDULE_FIELDS & set(payload)) or (task.enabled and not was_enabled):
|
||
hours = schedule.parse_hours(task.schedule_hours)
|
||
days = schedule.parse_days(task.schedule_days)
|
||
_require_clock_fields(task.schedule_mode, hours, days)
|
||
task.next_run_at = schedule.next_occurrence(
|
||
mode=task.schedule_mode,
|
||
interval_minutes=task.interval_minutes,
|
||
hours=hours,
|
||
days=days,
|
||
minute=task.schedule_minute,
|
||
after_ms=get_current_timestamp(),
|
||
)
|
||
|
||
task.updated_at = get_current_timestamp()
|
||
await session.flush()
|
||
return task
|
||
|
||
|
||
async def delete_task(session: AsyncSession, task_id: int) -> None:
|
||
task = await session.get(MonitorTask, task_id)
|
||
if task is None:
|
||
raise ValueError(f"Task {task_id} not found")
|
||
await session.delete(task)
|
||
|
||
|
||
def trigger_manual_run(task_id: int) -> None:
|
||
"""Fire a run in the background and return immediately.
|
||
|
||
A crawl takes minutes, so the HTTP request must not wait for it. The UI
|
||
follows progress through the logs WebSocket and the run history.
|
||
"""
|
||
task = asyncio.create_task(execute_task(task_id, trigger="manual"))
|
||
_background_runs.add(task)
|
||
task.add_done_callback(_background_runs.discard)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Dashboard queries
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def _latest_successful_run_id(session: AsyncSession, task_id: int) -> Optional[int]:
|
||
return await session.scalar(
|
||
select(MonitorRun.id)
|
||
.where(
|
||
MonitorRun.task_id == task_id,
|
||
MonitorRun.status.in_((RUN_SUCCESS, RUN_PARTIAL)),
|
||
)
|
||
.order_by(MonitorRun.id.desc())
|
||
.limit(1)
|
||
)
|
||
|
||
|
||
def _delta(current: Optional[int], previous: Optional[int]) -> Optional[int]:
|
||
if current is None or previous is None:
|
||
return None
|
||
return current - previous
|
||
|
||
|
||
async def list_notes(
|
||
session: AsyncSession,
|
||
task_id: Optional[int] = None,
|
||
only_new: bool = False,
|
||
limit: int = 200,
|
||
platform: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""Tracked notes with their latest metrics and change vs the previous run."""
|
||
query = select(MonitorNote).order_by(MonitorNote.last_seen_at.desc()).limit(limit)
|
||
if task_id is not None:
|
||
query = query.where(MonitorNote.task_id == task_id)
|
||
if platform is not None:
|
||
scoped = await platform_task_ids(session, platform)
|
||
if not scoped:
|
||
return []
|
||
query = query.where(MonitorNote.task_id.in_(scoped))
|
||
|
||
notes = list((await session.scalars(query)).all())
|
||
if not notes:
|
||
return []
|
||
|
||
note_ids = [note.note_id for note in notes]
|
||
|
||
# Fetch every snapshot for these notes in one go and gather the two most
|
||
# recent per note, rather than issuing two queries per note.
|
||
snapshots = list(
|
||
(
|
||
await session.scalars(
|
||
select(MonitorNoteMetric)
|
||
.where(MonitorNoteMetric.note_id.in_(note_ids))
|
||
.order_by(MonitorNoteMetric.note_id, MonitorNoteMetric.run_id.desc())
|
||
)
|
||
).all()
|
||
)
|
||
by_note: Dict[str, List[MonitorNoteMetric]] = {}
|
||
for snapshot in snapshots:
|
||
by_note.setdefault(snapshot.note_id, []).append(snapshot)
|
||
|
||
latest_run_ids: Dict[int, Optional[int]] = {}
|
||
result: List[Dict[str, Any]] = []
|
||
|
||
for note in notes:
|
||
series = by_note.get(note.note_id, [])
|
||
current = series[0] if series else None
|
||
previous = series[1] if len(series) > 1 else None
|
||
|
||
if only_new:
|
||
if note.task_id not in latest_run_ids:
|
||
latest_run_ids[note.task_id] = await _latest_successful_run_id(session, note.task_id)
|
||
if note.first_seen_run_id != latest_run_ids[note.task_id]:
|
||
continue
|
||
|
||
result.append(
|
||
{
|
||
"task_id": note.task_id,
|
||
"note_id": note.note_id,
|
||
"title": note.title,
|
||
"note_url": note.note_url,
|
||
# 按博主分组用。creator_hash 是唯一稳定的创作者标识（原始 user_id
|
||
# 被爬虫刻意匿名化了），creator_name 是已脱敏的昵称。
|
||
"creator_hash": note.creator_hash,
|
||
"creator_name": note.creator_name,
|
||
# 优先给本地缓存地址：远程地址带签名、会过期（实测隔天即 403），
|
||
# 本地那份不会。没有缓存时才退回远程，至少让图先显示出来。
|
||
"cover": covers.cover_url(note.note_id, note.cover),
|
||
"first_seen_at": note.first_seen_at,
|
||
"last_seen_at": note.last_seen_at,
|
||
"is_new": note.first_seen_run_id == latest_run_ids.get(note.task_id),
|
||
"metrics": {
|
||
"liked_count": current.liked_count if current else None,
|
||
"comment_count": current.comment_count if current else None,
|
||
"collected_count": current.collected_count if current else None,
|
||
"share_count": current.share_count if current else None,
|
||
},
|
||
"deltas": {
|
||
"liked_count": _delta(
|
||
current.liked_count if current else None,
|
||
previous.liked_count if previous else None,
|
||
),
|
||
"comment_count": _delta(
|
||
current.comment_count if current else None,
|
||
previous.comment_count if previous else None,
|
||
),
|
||
"collected_count": _delta(
|
||
current.collected_count if current else None,
|
||
previous.collected_count if previous else None,
|
||
),
|
||
"share_count": _delta(
|
||
current.share_count if current else None,
|
||
previous.share_count if previous else None,
|
||
),
|
||
},
|
||
"snapshot_count": len(series),
|
||
}
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
async def note_series(session: AsyncSession, note_id: str, task_id: Optional[int] = None) -> List[Dict[str, Any]]:
|
||
"""Metric time series for one note."""
|
||
query = (
|
||
select(MonitorNoteMetric)
|
||
.where(MonitorNoteMetric.note_id == note_id)
|
||
.order_by(MonitorNoteMetric.run_id)
|
||
)
|
||
if task_id is not None:
|
||
query = query.where(MonitorNoteMetric.task_id == task_id)
|
||
|
||
return [
|
||
{
|
||
"run_id": row.run_id,
|
||
"captured_at": row.captured_at,
|
||
"liked_count": row.liked_count,
|
||
"comment_count": row.comment_count,
|
||
"collected_count": row.collected_count,
|
||
"share_count": row.share_count,
|
||
}
|
||
for row in (await session.scalars(query)).all()
|
||
]
|
||
|
||
|
||
async def _note_meta_map(
|
||
session: AsyncSession, note_ids: List[str]
|
||
) -> Dict[str, Dict[str, Any]]:
|
||
"""Look up note title/cover/url for a set of note ids.
|
||
|
||
Fetched as one query and joined in Python rather than as a SQL join: the
|
||
comment table has no foreign key to the note table (both are keyed by the
|
||
platform's note id, per task), and a single IN() is easier to follow here.
|
||
"""
|
||
if not note_ids:
|
||
return {}
|
||
|
||
rows = (
|
||
await session.scalars(select(MonitorNote).where(MonitorNote.note_id.in_(set(note_ids))))
|
||
).all()
|
||
return {
|
||
row.note_id: {
|
||
"note_title": row.title,
|
||
"note_cover": covers.cover_url(row.note_id, row.cover),
|
||
"note_url": row.note_url,
|
||
"task_id": row.task_id,
|
||
# 博主维度也带上，评论流才能按 博主 -> 作品 -> 评论 三级展开。
|
||
"creator_hash": row.creator_hash,
|
||
"creator_name": row.creator_name,
|
||
}
|
||
for row in rows
|
||
}
|
||
|
||
|
||
async def list_comments(
|
||
session: AsyncSession,
|
||
task_id: Optional[int] = None,
|
||
note_id: Optional[str] = None,
|
||
limit: int = 200,
|
||
platform: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""Comments, each carrying the note it belongs to.
|
||
|
||
The note association is the point: without it a comment stream is unreadable,
|
||
since a bare note_id tells the operator nothing.
|
||
"""
|
||
query = select(MonitorComment).order_by(MonitorComment.first_seen_at.desc()).limit(limit)
|
||
if task_id is not None:
|
||
query = query.where(MonitorComment.task_id == task_id)
|
||
if note_id is not None:
|
||
query = query.where(MonitorComment.note_id == note_id)
|
||
if platform is not None:
|
||
scoped = await platform_task_ids(session, platform)
|
||
if not scoped:
|
||
return []
|
||
query = query.where(MonitorComment.task_id.in_(scoped))
|
||
|
||
comments = list((await session.scalars(query)).all())
|
||
meta = await _note_meta_map(session, [row.note_id for row in comments])
|
||
|
||
return [
|
||
{
|
||
"task_id": row.task_id,
|
||
"note_id": row.note_id,
|
||
"comment_id": row.comment_id,
|
||
"content": row.content,
|
||
"nickname": row.nickname,
|
||
"create_time": row.create_time,
|
||
"like_count": row.like_count,
|
||
"sub_comment_count": row.sub_comment_count,
|
||
"first_seen_at": row.first_seen_at,
|
||
"note_title": meta.get(row.note_id, {}).get("note_title", ""),
|
||
"note_cover": meta.get(row.note_id, {}).get("note_cover", ""),
|
||
"note_url": meta.get(row.note_id, {}).get("note_url", ""),
|
||
"note_creator_hash": meta.get(row.note_id, {}).get("creator_hash", ""),
|
||
"note_creator_name": meta.get(row.note_id, {}).get("creator_name", ""),
|
||
}
|
||
for row in comments
|
||
]
|
||
|
||
|
||
async def comment_note_groups(
|
||
session: AsyncSession, task_id: Optional[int] = None, platform: Optional[str] = None
|
||
) -> List[Dict[str, Any]]:
|
||
"""Notes that have comments, newest first, with their comment counts.
|
||
|
||
Feeds the comment filter dropdown: the operator picks a work by title, so
|
||
the counts need to be visible before choosing.
|
||
"""
|
||
scoped_ids: Optional[List[int]] = None
|
||
if platform is not None:
|
||
scoped_ids = await platform_task_ids(session, platform)
|
||
if not scoped_ids:
|
||
return []
|
||
|
||
count_query = select(
|
||
MonitorComment.note_id, func.count().label("comment_count")
|
||
).group_by(MonitorComment.note_id)
|
||
if task_id is not None:
|
||
count_query = count_query.where(MonitorComment.task_id == task_id)
|
||
if scoped_ids is not None:
|
||
count_query = count_query.where(MonitorComment.task_id.in_(scoped_ids))
|
||
|
||
counts = {row.note_id: row.comment_count for row in (await session.execute(count_query)).all()}
|
||
if not counts:
|
||
return []
|
||
|
||
latest_query = (
|
||
select(MonitorComment.note_id, func.max(MonitorComment.first_seen_at).label("latest"))
|
||
.where(MonitorComment.note_id.in_(set(counts)))
|
||
.group_by(MonitorComment.note_id)
|
||
)
|
||
if task_id is not None:
|
||
latest_query = latest_query.where(MonitorComment.task_id == task_id)
|
||
if scoped_ids is not None:
|
||
latest_query = latest_query.where(MonitorComment.task_id.in_(scoped_ids))
|
||
latest = {row.note_id: row.latest for row in (await session.execute(latest_query)).all()}
|
||
|
||
meta = await _note_meta_map(session, list(counts))
|
||
|
||
groups = [
|
||
{
|
||
"note_id": note_id,
|
||
"note_title": meta.get(note_id, {}).get("note_title", ""),
|
||
"note_cover": meta.get(note_id, {}).get("note_cover", ""),
|
||
"note_url": meta.get(note_id, {}).get("note_url", ""),
|
||
"creator_hash": meta.get(note_id, {}).get("creator_hash", ""),
|
||
"creator_name": meta.get(note_id, {}).get("creator_name", ""),
|
||
"comment_count": count,
|
||
"latest_at": latest.get(note_id, 0),
|
||
}
|
||
for note_id, count in counts.items()
|
||
]
|
||
groups.sort(key=lambda group: group["latest_at"], reverse=True)
|
||
return groups
|
||
|
||
|
||
async def list_events(
|
||
session: AsyncSession,
|
||
task_id: Optional[int] = None,
|
||
event_type: Optional[str] = None,
|
||
since_id: Optional[int] = None,
|
||
limit: int = 200,
|
||
platform: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
query = select(MonitorEvent).order_by(MonitorEvent.id.desc()).limit(limit)
|
||
if task_id is not None:
|
||
query = query.where(MonitorEvent.task_id == task_id)
|
||
if event_type is not None:
|
||
query = query.where(MonitorEvent.type == event_type)
|
||
if since_id is not None:
|
||
query = query.where(MonitorEvent.id > since_id)
|
||
if platform is not None:
|
||
scoped = await platform_task_ids(session, platform)
|
||
if not scoped:
|
||
return []
|
||
query = query.where(MonitorEvent.task_id.in_(scoped))
|
||
|
||
return [
|
||
{
|
||
"id": row.id,
|
||
"task_id": row.task_id,
|
||
"run_id": row.run_id,
|
||
"type": row.type,
|
||
"severity": row.severity,
|
||
"target_kind": row.target_kind,
|
||
"target_id": row.target_id,
|
||
"title": row.title,
|
||
"created_at": row.created_at,
|
||
"is_read": row.is_read,
|
||
}
|
||
for row in (await session.scalars(query)).all()
|
||
]
|
||
|
||
|
||
async def list_runs(session: AsyncSession, task_id: int, limit: int = 50) -> List[Dict[str, Any]]:
|
||
rows = (
|
||
await session.scalars(
|
||
select(MonitorRun)
|
||
.where(MonitorRun.task_id == task_id)
|
||
.order_by(MonitorRun.id.desc())
|
||
.limit(limit)
|
||
)
|
||
).all()
|
||
|
||
return [
|
||
{
|
||
"id": row.id,
|
||
"task_id": row.task_id,
|
||
"status": row.status,
|
||
"trigger": row.trigger,
|
||
"queued_at": row.queued_at,
|
||
"started_at": row.started_at,
|
||
"finished_at": row.finished_at,
|
||
"exit_code": row.exit_code,
|
||
"notes_fetched": row.notes_fetched,
|
||
"comments_fetched": row.comments_fetched,
|
||
"new_notes": row.new_notes,
|
||
"new_comments": row.new_comments,
|
||
"is_baseline": row.is_baseline,
|
||
"max_comments_count": row.max_comments_count,
|
||
"error_message": row.error_message,
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
|
||
async def list_tasks(
|
||
session: AsyncSession, platform: Optional[str] = None
|
||
) -> List[Dict[str, Any]]:
|
||
query = select(MonitorTask).order_by(MonitorTask.id)
|
||
if platform is not None:
|
||
query = query.where(MonitorTask.platform == platform)
|
||
|
||
tasks = list((await session.scalars(query)).all())
|
||
if not tasks:
|
||
return []
|
||
|
||
counts = dict(
|
||
(
|
||
await session.execute(
|
||
select(MonitorTarget.task_id, func.count())
|
||
.group_by(MonitorTarget.task_id)
|
||
)
|
||
).all()
|
||
)
|
||
unread = dict(
|
||
(
|
||
await session.execute(
|
||
select(MonitorEvent.task_id, func.count())
|
||
.where(MonitorEvent.is_read.is_(False))
|
||
.group_by(MonitorEvent.task_id)
|
||
)
|
||
).all()
|
||
)
|
||
|
||
return [
|
||
{
|
||
"id": task.id,
|
||
"name": task.name,
|
||
"platform": task.platform,
|
||
"mode": task.mode,
|
||
"enabled": task.enabled,
|
||
"interval_minutes": task.interval_minutes,
|
||
**_schedule_fields(task),
|
||
"max_notes_count": task.max_notes_count,
|
||
"enable_comments": task.enable_comments,
|
||
"max_comments_count": task.max_comments_count,
|
||
"run_timeout_seconds": task.run_timeout_seconds,
|
||
"notify_enabled": task.notify_enabled,
|
||
"notify_failures": task.notify_failures,
|
||
"next_run_at": task.next_run_at,
|
||
"last_run_at": task.last_run_at,
|
||
"last_status": task.last_status,
|
||
"last_error": task.last_error,
|
||
"last_notified_at": task.last_notified_at,
|
||
"target_count": counts.get(task.id, 0),
|
||
"targets": [
|
||
{"id": t.id, "external_id": t.external_id, "raw_value": t.raw_value, "enabled": t.enabled}
|
||
for t in task.targets
|
||
],
|
||
"unread_events": unread.get(task.id, 0),
|
||
}
|
||
for task in tasks
|
||
]
|
||
|
||
|
||
def _schedule_fields(task: MonitorTask) -> Dict[str, Any]:
|
||
"""The schedule columns, plus the sentence the task list renders.
|
||
|
||
The label is composed here rather than in the frontend so the list and the
|
||
editor cannot drift on what a given schedule means.
|
||
"""
|
||
hours = schedule.parse_hours(task.schedule_hours)
|
||
days = schedule.parse_days(task.schedule_days)
|
||
return {
|
||
"schedule_mode": task.schedule_mode,
|
||
"schedule_hours": hours,
|
||
"schedule_days": days,
|
||
"schedule_minute": task.schedule_minute,
|
||
"schedule_label": schedule.describe(
|
||
mode=task.schedule_mode,
|
||
interval_minutes=task.interval_minutes,
|
||
hours=hours,
|
||
days=days,
|
||
minute=task.schedule_minute,
|
||
),
|
||
}
|
||
|
||
|
||
async def overview(session: AsyncSession, platform: Optional[str] = None) -> Dict[str, Any]:
|
||
"""Headline numbers for the dashboard tiles, scoped to one platform."""
|
||
now = get_current_timestamp()
|
||
day_ago = now - 24 * 60 * 60 * 1000
|
||
|
||
# Nothing but the task table carries a platform column, so the other counts
|
||
# are scoped through the platform's task ids.
|
||
scoped: Optional[List[int]] = None
|
||
if platform is not None:
|
||
scoped = await platform_task_ids(session, platform)
|
||
|
||
def by_task(stmt, column):
|
||
return stmt if scoped is None else stmt.where(column.in_(scoped))
|
||
|
||
task_count = select(func.count()).select_from(MonitorTask)
|
||
if platform is not None:
|
||
task_count = task_count.where(MonitorTask.platform == platform)
|
||
|
||
enabled_count = select(func.count()).select_from(MonitorTask).where(
|
||
MonitorTask.enabled.is_(True)
|
||
)
|
||
if platform is not None:
|
||
enabled_count = enabled_count.where(MonitorTask.platform == platform)
|
||
|
||
return {
|
||
"platform": platform,
|
||
"tasks": await session.scalar(task_count) or 0,
|
||
"enabled_tasks": await session.scalar(enabled_count) or 0,
|
||
"notes": await session.scalar(
|
||
by_task(select(func.count()).select_from(MonitorNote), MonitorNote.task_id)
|
||
)
|
||
or 0,
|
||
"comments": await session.scalar(
|
||
by_task(select(func.count()).select_from(MonitorComment), MonitorComment.task_id)
|
||
)
|
||
or 0,
|
||
"events_24h": await session.scalar(
|
||
by_task(
|
||
select(func.count())
|
||
.select_from(MonitorEvent)
|
||
.where(MonitorEvent.created_at >= day_ago),
|
||
MonitorEvent.task_id,
|
||
)
|
||
)
|
||
or 0,
|
||
"unread_events": await session.scalar(
|
||
by_task(
|
||
select(func.count())
|
||
.select_from(MonitorEvent)
|
||
.where(MonitorEvent.is_read.is_(False)),
|
||
MonitorEvent.task_id,
|
||
)
|
||
)
|
||
or 0,
|
||
"running_runs": await session.scalar(
|
||
by_task(
|
||
select(func.count())
|
||
.select_from(MonitorRun)
|
||
.where(MonitorRun.status == "running"),
|
||
MonitorRun.task_id,
|
||
)
|
||
)
|
||
or 0,
|
||
}
|
||
|
||
|
||
async def mark_events_read(session: AsyncSession, task_id: Optional[int] = None) -> int:
|
||
query = select(MonitorEvent).where(MonitorEvent.is_read.is_(False))
|
||
if task_id is not None:
|
||
query = query.where(MonitorEvent.task_id == task_id)
|
||
rows = list((await session.scalars(query)).all())
|
||
for row in rows:
|
||
row.is_read = True
|
||
return len(rows)
|