670 lines
27 KiB
Python
670 lines
27 KiB
Python
"""Scheduled task orchestration service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import mimetypes
|
|
import re
|
|
import shutil
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from croniter import croniter
|
|
from langgraph.checkpoint.base import empty_checkpoint
|
|
|
|
from deerflow.config.paths import get_paths
|
|
from deerflow.persistence.scheduled_tasks import ScheduledTaskStore
|
|
from deerflow.runtime import serialize_channel_values
|
|
from deerflow.runtime.scheduler.element import ElementDeliveryClient
|
|
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
|
from deerflow.utils.time import now_iso
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class _RuntimeUser:
|
|
id: str
|
|
|
|
|
|
def _coerce_datetime(value: datetime | str | None) -> datetime | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
dt = value
|
|
else:
|
|
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=UTC)
|
|
return dt.astimezone(UTC)
|
|
|
|
|
|
def _normalize_timezone(timezone: str | None) -> str:
|
|
tz = (timezone or "UTC").strip() or "UTC"
|
|
try:
|
|
ZoneInfo(tz)
|
|
return tz
|
|
except Exception:
|
|
return "UTC"
|
|
|
|
|
|
def _schedule_to_cron(schedule: str) -> str:
|
|
value = schedule.strip()
|
|
if re.fullmatch(r"(\S+\s+){4}\S+", value):
|
|
return value
|
|
|
|
lower = value.lower()
|
|
hour_min = re.search(r"(?:(?:at|@)\s*)?(\d{1,2})(?::(\d{2}))?\s*(am|pm)?", lower)
|
|
cn_hour_min = re.search(r"(\d{1,2})\s*[点時时](?:(\d{1,2})\s*分?)?", value)
|
|
|
|
hour = 9
|
|
minute = 0
|
|
ampm = None
|
|
if hour_min:
|
|
hour = int(hour_min.group(1))
|
|
minute = int(hour_min.group(2) or 0)
|
|
ampm = hour_min.group(3)
|
|
elif cn_hour_min:
|
|
hour = int(cn_hour_min.group(1))
|
|
minute = int(cn_hour_min.group(2) or 0)
|
|
|
|
if ampm == "pm" and hour < 12:
|
|
hour += 12
|
|
if ampm == "am" and hour == 12:
|
|
hour = 0
|
|
hour = max(0, min(23, hour))
|
|
minute = max(0, min(59, minute))
|
|
|
|
if any(token in lower for token in ("weekday", "weekdays", "workday")) or "工作日" in value:
|
|
return f"{minute} {hour} * * 1-5"
|
|
|
|
weekdays = {
|
|
"monday": 1,
|
|
"tuesday": 2,
|
|
"wednesday": 3,
|
|
"thursday": 4,
|
|
"friday": 5,
|
|
"saturday": 6,
|
|
"sunday": 0,
|
|
"周一": 1,
|
|
"星期一": 1,
|
|
"周二": 2,
|
|
"星期二": 2,
|
|
"周三": 3,
|
|
"星期三": 3,
|
|
"周四": 4,
|
|
"星期四": 4,
|
|
"周五": 5,
|
|
"星期五": 5,
|
|
"周六": 6,
|
|
"星期六": 6,
|
|
"周日": 0,
|
|
"周天": 0,
|
|
"星期日": 0,
|
|
"星期天": 0,
|
|
}
|
|
for token, weekday in weekdays.items():
|
|
if token in lower or token in value:
|
|
return f"{minute} {hour} * * {weekday}"
|
|
|
|
every_hours = re.search(r"every\s+(\d+)\s*h", lower)
|
|
cn_every_hours = re.search(r"每隔\s*(\d+)\s*小?时", value)
|
|
if every_hours or cn_every_hours:
|
|
hours = int((every_hours or cn_every_hours).group(1))
|
|
hours = max(1, min(23, hours))
|
|
return f"0 */{hours} * * *"
|
|
|
|
return f"{minute} {hour} * * *"
|
|
|
|
|
|
def _next_run_at(cron_expr: str, timezone: str, base: datetime | None = None) -> datetime:
|
|
tz = ZoneInfo(timezone)
|
|
base_local = (base or datetime.now(UTC)).astimezone(tz)
|
|
next_local = croniter(cron_expr, base_local).get_next(datetime)
|
|
if next_local.tzinfo is None:
|
|
next_local = next_local.replace(tzinfo=tz)
|
|
return next_local.astimezone(UTC)
|
|
|
|
|
|
def _looks_like_markdown(content: str) -> bool:
|
|
markdown_patterns = (
|
|
r"^#{1,6}\s+",
|
|
r"^\s*[-*+]\s+",
|
|
r"^\s*\d+\.\s+",
|
|
r"```",
|
|
r"\|.+\|",
|
|
r"\[[^\]]+\]\([^)]+\)",
|
|
)
|
|
return any(re.search(pattern, content, flags=re.MULTILINE) for pattern in markdown_patterns)
|
|
|
|
|
|
def _build_result(
|
|
task: dict[str, Any],
|
|
*,
|
|
scheduled_for: datetime,
|
|
content: str | None,
|
|
files: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
result_content = content or f"Scheduled task completed: {task['name']}"
|
|
task_tz = ZoneInfo(task.get("timezone") or "UTC")
|
|
local_time = scheduled_for.astimezone(task_tz)
|
|
return {
|
|
"title": f"{task['name']} - {local_time.strftime('%Y-%m-%d %H:%M')}",
|
|
"content": result_content,
|
|
"format": "markdown" if _looks_like_markdown(result_content) else "text",
|
|
"files": files or [],
|
|
"created_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
|
|
|
|
class ScheduledTaskService:
|
|
def __init__(self, app: Any, store: ScheduledTaskStore, *, poll_interval_seconds: float = 30.0) -> None:
|
|
self.app = app
|
|
self.store = store
|
|
self.poll_interval_seconds = poll_interval_seconds
|
|
self._loop_task: asyncio.Task | None = None
|
|
self._stopping = asyncio.Event()
|
|
self._running_task_ids: set[str] = set()
|
|
# All scheduled jobs deliver into one per-user scheduler thread. The
|
|
# current RunManager rejects concurrent runs on the same thread, so
|
|
# serialize starts here instead of letting reports fail with conflicts.
|
|
self._agent_run_lock = asyncio.Lock()
|
|
self._element_client = ElementDeliveryClient.from_env()
|
|
|
|
async def start(self) -> None:
|
|
if self._loop_task is None or self._loop_task.done():
|
|
self._stopping.clear()
|
|
self._loop_task = asyncio.create_task(self._run_loop(), name="scheduled-task-loop")
|
|
|
|
async def stop(self) -> None:
|
|
self._stopping.set()
|
|
if self._loop_task is not None:
|
|
self._loop_task.cancel()
|
|
try:
|
|
await self._loop_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def ensure_scheduler_thread(self, user_id: str) -> dict[str, Any]:
|
|
thread = await self.store.get_or_create_scheduler_thread(user_id, lambda: str(uuid.uuid4()))
|
|
await self._ensure_thread_meta(user_id, thread["thread_id"])
|
|
return thread
|
|
|
|
async def list_tasks(self, user_id: str) -> list[dict[str, Any]]:
|
|
return await self.store.list_tasks(user_id)
|
|
|
|
async def list_runs(self, user_id: str, task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
|
|
task = await self.store.get_task(task_id, user_id)
|
|
if task is None:
|
|
return []
|
|
return await self.store.list_runs(task_id, user_id, limit=limit)
|
|
|
|
async def get_run(self, user_id: str, run_id: str) -> dict[str, Any] | None:
|
|
return await self.store.get_run(run_id, user_id)
|
|
|
|
async def delete_run(self, user_id: str, run_id: str) -> bool:
|
|
deleted = await self.store.delete_run(run_id, user_id)
|
|
if deleted:
|
|
archive_dir = get_paths().base_dir / "scheduled-task-runs" / run_id
|
|
if archive_dir.exists():
|
|
try:
|
|
shutil.rmtree(archive_dir)
|
|
except Exception:
|
|
logger.warning("Failed to delete scheduled task run archive %s", archive_dir, exc_info=True)
|
|
return deleted
|
|
|
|
async def get_delivery_profile(self, user_id: str) -> dict[str, Any]:
|
|
return await self.store.get_delivery_profile(user_id)
|
|
|
|
async def update_delivery_profile(self, user_id: str, data: dict[str, Any]) -> dict[str, Any]:
|
|
return await self.store.update_delivery_profile(user_id, data)
|
|
|
|
async def create_task(
|
|
self,
|
|
*,
|
|
user_id: str,
|
|
name: str,
|
|
prompt: str,
|
|
schedule: str,
|
|
timezone: str | None = None,
|
|
source_thread_id: str | None = None,
|
|
source_agent_name: str | None = None,
|
|
execution_agent_name: str | None = None,
|
|
execution_context: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
tz = _normalize_timezone(timezone)
|
|
cron_expr = _schedule_to_cron(schedule)
|
|
next_run_at = _next_run_at(cron_expr, tz)
|
|
scheduler_thread = await self.ensure_scheduler_thread(user_id)
|
|
context = dict(execution_context or {})
|
|
if execution_agent_name:
|
|
context["agent_name"] = execution_agent_name
|
|
task = await self.store.create_task(
|
|
{
|
|
"task_id": str(uuid.uuid4()),
|
|
"user_id": user_id,
|
|
"scheduler_thread_id": scheduler_thread["thread_id"],
|
|
"source_thread_id": source_thread_id,
|
|
"source_agent_name": source_agent_name,
|
|
"execution_agent_name": execution_agent_name,
|
|
"name": name.strip() or "Scheduled task",
|
|
"prompt": prompt.strip(),
|
|
"schedule_text": schedule.strip(),
|
|
"cron_expr": cron_expr,
|
|
"timezone": tz,
|
|
"enabled": True,
|
|
"next_run_at": next_run_at,
|
|
"execution_context": context,
|
|
}
|
|
)
|
|
return task
|
|
|
|
async def update_task(self, user_id: str, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
|
|
update_data: dict[str, Any] = {}
|
|
for key in ("name", "prompt", "enabled", "execution_agent_name"):
|
|
if key in data and data[key] is not None:
|
|
update_data[key] = data[key]
|
|
timezone = data.get("timezone")
|
|
if timezone is not None:
|
|
update_data["timezone"] = _normalize_timezone(timezone)
|
|
if data.get("schedule") is not None:
|
|
schedule = str(data["schedule"])
|
|
cron_expr = _schedule_to_cron(schedule)
|
|
tz = update_data.get("timezone")
|
|
if tz is None:
|
|
existing = await self.store.get_task(task_id, user_id)
|
|
tz = (existing or {}).get("timezone", "UTC")
|
|
update_data["schedule_text"] = schedule
|
|
update_data["cron_expr"] = cron_expr
|
|
update_data["next_run_at"] = _next_run_at(cron_expr, tz)
|
|
if data.get("execution_context") is not None:
|
|
update_data["execution_context"] = data["execution_context"]
|
|
return await self.store.update_task(task_id, user_id, update_data)
|
|
|
|
async def delete_task(self, user_id: str, task_id: str) -> bool:
|
|
return await self.store.delete_task(task_id, user_id)
|
|
|
|
async def pause_task(self, user_id: str, task_id: str) -> dict[str, Any] | None:
|
|
return await self.store.update_task(task_id, user_id, {"enabled": False})
|
|
|
|
async def resume_task(self, user_id: str, task_id: str) -> dict[str, Any] | None:
|
|
existing = await self.store.get_task(task_id, user_id)
|
|
if existing is None:
|
|
return None
|
|
next_run_at = _next_run_at(existing["cron_expr"], existing.get("timezone") or "UTC")
|
|
return await self.store.update_task(task_id, user_id, {"enabled": True, "next_run_at": next_run_at})
|
|
|
|
async def trigger_task(self, user_id: str, task_id: str) -> dict[str, Any] | None:
|
|
task = await self.store.get_task(task_id, user_id)
|
|
if task is None:
|
|
return None
|
|
asyncio.create_task(self._execute_task(task, scheduled_for=datetime.now(UTC), manual=True))
|
|
return task
|
|
|
|
async def _run_loop(self) -> None:
|
|
while not self._stopping.is_set():
|
|
try:
|
|
due = await self.store.due_tasks(datetime.now(UTC), limit=20)
|
|
for task in due:
|
|
task_id = task["task_id"]
|
|
if task_id in self._running_task_ids:
|
|
continue
|
|
self._running_task_ids.add(task_id)
|
|
asyncio.create_task(self._execute_and_untrack(task))
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("Scheduled task scan failed")
|
|
try:
|
|
await asyncio.wait_for(self._stopping.wait(), timeout=self.poll_interval_seconds)
|
|
except TimeoutError:
|
|
pass
|
|
|
|
async def _execute_and_untrack(self, task: dict[str, Any]) -> None:
|
|
try:
|
|
scheduled_for = _coerce_datetime(task.get("next_run_at")) or datetime.now(UTC)
|
|
await self._execute_task(task, scheduled_for=scheduled_for, manual=False)
|
|
finally:
|
|
self._running_task_ids.discard(task["task_id"])
|
|
|
|
async def _execute_task(self, task: dict[str, Any], *, scheduled_for: datetime, manual: bool) -> None:
|
|
run_row = await self.store.create_run_once(
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"task_id": task["task_id"],
|
|
"user_id": task["user_id"],
|
|
"scheduler_thread_id": task["scheduler_thread_id"],
|
|
"scheduled_for": scheduled_for,
|
|
"status": "running",
|
|
"started_at": datetime.now(UTC),
|
|
}
|
|
)
|
|
if run_row is None:
|
|
return
|
|
|
|
if not manual:
|
|
next_run_at = _next_run_at(task["cron_expr"], task.get("timezone") or "UTC", base=scheduled_for)
|
|
await self.store.update_task(task["task_id"], task["user_id"], {"next_run_at": next_run_at, "last_status": "running"})
|
|
|
|
try:
|
|
await self._ensure_thread_meta(task["user_id"], task["scheduler_thread_id"])
|
|
async with self._agent_run_lock:
|
|
output_snapshot = self._snapshot_result_files(task["scheduler_thread_id"], task["user_id"])
|
|
record = await self._start_agent_run(task, scheduled_for=scheduled_for)
|
|
await self.store.update_run(run_row["id"], {"agent_run_id": record.run_id})
|
|
if record.task is not None:
|
|
await record.task
|
|
result_text = await self._extract_latest_ai_text(task["scheduler_thread_id"])
|
|
result_files = await self._extract_result_files(
|
|
task["scheduler_thread_id"],
|
|
task["user_id"],
|
|
run_id=run_row["id"],
|
|
output_snapshot=output_snapshot,
|
|
)
|
|
result = _build_result(task, scheduled_for=scheduled_for, content=result_text, files=result_files)
|
|
await self.store.update_run(
|
|
run_row["id"],
|
|
{"status": "succeeded", "finished_at": datetime.now(UTC), "result": result},
|
|
)
|
|
await self.store.update_task(task["task_id"], task["user_id"], {"last_run_at": datetime.now(UTC), "last_status": "succeeded"})
|
|
await self._deliver_element_result(task, result["content"])
|
|
except Exception as exc:
|
|
logger.exception("Scheduled task %s failed", task["task_id"])
|
|
result = _build_result(
|
|
task,
|
|
scheduled_for=scheduled_for,
|
|
content=f"Scheduled task failed: {exc}",
|
|
)
|
|
await self.store.update_run(
|
|
run_row["id"],
|
|
{"status": "failed", "finished_at": datetime.now(UTC), "error": str(exc), "result": result},
|
|
)
|
|
await self.store.update_task(task["task_id"], task["user_id"], {"last_run_at": datetime.now(UTC), "last_status": "failed"})
|
|
|
|
async def _ensure_thread_meta(self, user_id: str, thread_id: str) -> None:
|
|
token = set_current_user(_RuntimeUser(id=user_id))
|
|
try:
|
|
thread_store = self.app.state.thread_store
|
|
existing = await thread_store.get(thread_id)
|
|
if existing is None:
|
|
await thread_store.create(
|
|
thread_id,
|
|
assistant_id="lead_agent",
|
|
display_name="定时任务助手",
|
|
metadata={"thread_type": "scheduler", "system": True},
|
|
)
|
|
checkpointer = self.app.state.checkpointer
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
await checkpointer.aput(
|
|
config,
|
|
empty_checkpoint(),
|
|
{
|
|
"step": -1,
|
|
"source": "input",
|
|
"writes": None,
|
|
"parents": {},
|
|
"created_at": now_iso(),
|
|
"thread_type": "scheduler",
|
|
},
|
|
{},
|
|
)
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
async def _start_agent_run(self, task: dict[str, Any], *, scheduled_for: datetime):
|
|
from app.gateway.routers.thread_runs import RunCreateRequest
|
|
from app.gateway.services import start_run
|
|
|
|
token = set_current_user(_RuntimeUser(id=task["user_id"]))
|
|
try:
|
|
request = SimpleNamespace(app=self.app)
|
|
execution_context = dict(task.get("execution_context") or {})
|
|
if task.get("execution_agent_name"):
|
|
execution_context["agent_name"] = task["execution_agent_name"]
|
|
execution_context.update(
|
|
{
|
|
"is_scheduled_run": True,
|
|
"thread_id": task["scheduler_thread_id"],
|
|
"thinking_enabled": execution_context.get("thinking_enabled", True),
|
|
"is_plan_mode": execution_context.get("is_plan_mode", True),
|
|
}
|
|
)
|
|
content = (
|
|
f"【定时任务触发】{task['name']}\n"
|
|
f"计划时间：{scheduled_for.astimezone(ZoneInfo(task.get('timezone') or 'UTC')).isoformat()}\n\n"
|
|
f"{task['prompt']}"
|
|
)
|
|
body = RunCreateRequest(
|
|
assistant_id="lead_agent",
|
|
input={"messages": [{"type": "human", "content": content}]},
|
|
metadata={"scheduled_task_id": task["task_id"], "scheduled_task_name": task["name"]},
|
|
context=execution_context,
|
|
multitask_strategy="enqueue" if False else "reject",
|
|
)
|
|
return await start_run(body, task["scheduler_thread_id"], request) # type: ignore[arg-type]
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
async def _extract_latest_ai_text(self, thread_id: str) -> str | None:
|
|
checkpointer = self.app.state.checkpointer
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
checkpoint_tuple = await checkpointer.aget_tuple(config)
|
|
if checkpoint_tuple is None:
|
|
return None
|
|
checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
|
|
channel_values = checkpoint.get("channel_values", {})
|
|
values = serialize_channel_values(channel_values)
|
|
messages = values.get("messages") or []
|
|
for message in reversed(messages):
|
|
if not isinstance(message, dict):
|
|
continue
|
|
message_type = message.get("type") or message.get("role")
|
|
if message_type not in {"ai", "assistant"}:
|
|
continue
|
|
content = message.get("content")
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts: list[str] = []
|
|
for part in content:
|
|
if isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str):
|
|
parts.append(part["text"])
|
|
if parts:
|
|
return "\n".join(parts)
|
|
return None
|
|
|
|
def _result_scan_dirs(self, thread_id: str, user_id: str) -> list[tuple[str, Path]]:
|
|
paths = get_paths()
|
|
return [
|
|
("/mnt/user-data/outputs", paths.sandbox_outputs_dir(thread_id, user_id=user_id)),
|
|
("/mnt/user-data/workspace", paths.sandbox_work_dir(thread_id, user_id=user_id)),
|
|
]
|
|
|
|
def _snapshot_result_files(self, thread_id: str, user_id: str) -> dict[str, tuple[int, int]]:
|
|
snapshot: dict[str, tuple[int, int]] = {}
|
|
for _, directory in self._result_scan_dirs(thread_id, user_id):
|
|
if not directory.exists() or not directory.is_dir():
|
|
continue
|
|
try:
|
|
for path in directory.rglob("*"):
|
|
if not path.is_file():
|
|
continue
|
|
stat = path.stat()
|
|
snapshot[str(path.resolve())] = (stat.st_mtime_ns, stat.st_size)
|
|
except Exception:
|
|
logger.debug("Failed to snapshot scheduled task result dir %s", directory, exc_info=True)
|
|
return snapshot
|
|
|
|
def _virtual_result_path(self, virtual_prefix: str, result_dir: Path, path: Path) -> str | None:
|
|
try:
|
|
relative = path.resolve().relative_to(result_dir.resolve())
|
|
except ValueError:
|
|
return None
|
|
return f"{virtual_prefix}/{relative.as_posix()}"
|
|
|
|
def _collect_changed_output_files(
|
|
self,
|
|
thread_id: str,
|
|
user_id: str,
|
|
output_snapshot: dict[str, tuple[int, int]],
|
|
) -> list[tuple[str, Path]]:
|
|
changed: list[tuple[str, Path]] = []
|
|
for virtual_prefix, directory in self._result_scan_dirs(thread_id, user_id):
|
|
if not directory.exists() or not directory.is_dir():
|
|
continue
|
|
try:
|
|
for path in sorted(directory.rglob("*"), key=lambda item: item.as_posix()):
|
|
if not path.is_file():
|
|
continue
|
|
resolved = path.resolve()
|
|
stat = path.stat()
|
|
previous = output_snapshot.get(str(resolved))
|
|
if previous == (stat.st_mtime_ns, stat.st_size):
|
|
continue
|
|
virtual_path = self._virtual_result_path(virtual_prefix, directory, resolved)
|
|
if virtual_path is not None:
|
|
changed.append((virtual_path, resolved))
|
|
except Exception:
|
|
logger.debug("Failed to collect changed scheduled task result files from %s", directory, exc_info=True)
|
|
return changed
|
|
|
|
def _is_in_result_scan_dirs(self, thread_id: str, user_id: str, path: Path) -> bool:
|
|
try:
|
|
resolved = path.resolve()
|
|
except Exception:
|
|
return False
|
|
for _, directory in self._result_scan_dirs(thread_id, user_id):
|
|
try:
|
|
resolved.relative_to(directory.resolve())
|
|
return True
|
|
except ValueError:
|
|
continue
|
|
return False
|
|
|
|
def _archive_result_file(self, run_id: str, path: Path) -> Path | None:
|
|
if not path.exists() or not path.is_file():
|
|
return None
|
|
|
|
archive_dir = get_paths().base_dir / "scheduled-task-runs" / run_id / "files"
|
|
archive_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
candidate = archive_dir / path.name
|
|
if candidate.exists():
|
|
stem = path.stem or "file"
|
|
suffix = path.suffix
|
|
index = 2
|
|
while candidate.exists():
|
|
candidate = archive_dir / f"{stem}-{index}{suffix}"
|
|
index += 1
|
|
|
|
shutil.copy2(path, candidate)
|
|
return candidate
|
|
|
|
def _build_file_result(self, *, virtual_path: str, source_path: Path, archive_path: Path) -> dict[str, Any]:
|
|
mime_type, _ = mimetypes.guess_type(archive_path.name)
|
|
suffix = archive_path.suffix.lower()
|
|
return {
|
|
"name": archive_path.name,
|
|
"path": str(archive_path),
|
|
"source_path": str(source_path),
|
|
"virtual_path": virtual_path,
|
|
"mime_type": mime_type or ("text/markdown" if suffix == ".md" else None),
|
|
"size": archive_path.stat().st_size if archive_path.exists() and archive_path.is_file() else None,
|
|
"kind": "markdown" if suffix == ".md" else "artifact",
|
|
}
|
|
|
|
def _is_new_or_changed_output_file(
|
|
self,
|
|
thread_id: str,
|
|
user_id: str,
|
|
path: Path,
|
|
output_snapshot: dict[str, tuple[int, int]],
|
|
) -> bool:
|
|
if not self._is_in_result_scan_dirs(thread_id, user_id, path):
|
|
return True
|
|
|
|
resolved = path.resolve()
|
|
if not resolved.exists() or not resolved.is_file():
|
|
return False
|
|
stat = resolved.stat()
|
|
return output_snapshot.get(str(resolved)) != (stat.st_mtime_ns, stat.st_size)
|
|
|
|
async def _extract_result_files(
|
|
self,
|
|
thread_id: str,
|
|
user_id: str,
|
|
*,
|
|
run_id: str,
|
|
output_snapshot: dict[str, tuple[int, int]] | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
checkpointer = self.app.state.checkpointer
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
checkpoint_tuple = await checkpointer.aget_tuple(config)
|
|
artifacts: list[Any] = []
|
|
if checkpoint_tuple is not None:
|
|
checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
|
|
channel_values = checkpoint.get("channel_values", {})
|
|
values = serialize_channel_values(channel_values)
|
|
raw_artifacts = values.get("artifacts") or []
|
|
if isinstance(raw_artifacts, list):
|
|
artifacts = raw_artifacts
|
|
|
|
result_files: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
|
|
def add_file(virtual_path: str, source_path: Path) -> None:
|
|
resolved_source = source_path.resolve()
|
|
seen_key = str(resolved_source)
|
|
if seen_key in seen:
|
|
return
|
|
seen.add(seen_key)
|
|
try:
|
|
archive_path = self._archive_result_file(run_id, resolved_source)
|
|
except Exception:
|
|
logger.debug("Failed to archive scheduled task artifact %s", resolved_source, exc_info=True)
|
|
return
|
|
if archive_path is None:
|
|
return
|
|
result_files.append(self._build_file_result(virtual_path=virtual_path, source_path=resolved_source, archive_path=archive_path))
|
|
|
|
for artifact in artifacts:
|
|
if not isinstance(artifact, str) or not artifact:
|
|
continue
|
|
try:
|
|
actual_path = get_paths().resolve_virtual_path(thread_id, artifact, user_id=user_id)
|
|
except TypeError:
|
|
actual_path = get_paths().resolve_virtual_path(thread_id, artifact)
|
|
except Exception:
|
|
logger.debug("Failed to resolve scheduled task artifact %s", artifact, exc_info=True)
|
|
continue
|
|
path = Path(actual_path)
|
|
if self._is_new_or_changed_output_file(thread_id, user_id, path, output_snapshot or {}):
|
|
add_file(artifact, path)
|
|
|
|
for virtual_path, path in self._collect_changed_output_files(thread_id, user_id, output_snapshot or {}):
|
|
add_file(virtual_path, path)
|
|
|
|
return result_files
|
|
|
|
async def _deliver_element_result(self, task: dict[str, Any], result_text: str) -> None:
|
|
if self._element_client is None:
|
|
return
|
|
profile = await self.store.get_delivery_profile(task["user_id"])
|
|
element_user_id = profile.get("element_user_id")
|
|
if not isinstance(element_user_id, str) or not element_user_id.strip():
|
|
return
|
|
existing_room_id: str | None = profile.get("element_room_id") or None
|
|
body = f"【定时任务】{task['name']}\n\n{result_text}"
|
|
try:
|
|
room_id = await self._element_client.send_direct_message(element_user_id.strip(), body, room_id=existing_room_id)
|
|
if room_id and room_id != existing_room_id:
|
|
await self.store.update_delivery_profile(task["user_id"], {"element_room_id": room_id})
|
|
except Exception:
|
|
logger.exception("Failed to deliver scheduled task %s to Element user %s", task["task_id"], element_user_id)
|