Daily Work MCP Server

from __future__ import annotations from datetime import date from typing import Sequence from sqlalchemy.orm import Session from app.core.config import get_settings from app.models import Tag, WorkEntry from app.services.repositories import ( EntryRepository, TagRepository, compute_date_range, normalize_tag_names, ) class EntryService: def __init__(self, session: Session): self.session = session self.settings = get_settings() self.entry_repo = EntryRepository(session) self.tag_repo = TagRepository(session) def _load_tags(self, tag_names: Sequence[str] | None) -> list[Tag]: normalized = normalize_tag_names(tag_names) if not normalized: return [] tag_map = self.tag_repo.fetch_by_names(normalized) missing = [name for name in normalized if name not in tag_map] if missing: raise ValueError(f"标签不存在: {', '.join(missing)}") return [tag_map[name] for name in normalized] def create_entry(self, description: str, tag_names: Sequence[str] | None = None) -> WorkEntry: tags = self._load_tags(tag_names) entry = self.entry_repo.create_entry(description=description, tags=tags) return entry def list_entries( self, *, range_type: str, target_date: date | None = None, tag_names: Sequence[str] | None = None, ) -> list[WorkEntry]: return self.entry_repo.list_entries( range_type=range_type, target_date=target_date, tag_names=tag_names, ) def export_entries( self, *, start_date: date | None = None, end_date: date | None = None, ) -> list[WorkEntry]: return self.entry_repo.export_entries(start_date=start_date, end_date=end_date) def search_entries( self, *, query: str, start_date: date | None = None, end_date: date | None = None, ) -> list[WorkEntry]: return self.entry_repo.search_entries( query=query, start_date=start_date, end_date=end_date, ) def summarize(self, *, days: int) -> list[dict[str, object]]: return self.entry_repo.summarize_last_days(days) def has_entry_today(self) -> bool: date_range = compute_date_range("daily", date.today()) return self.entry_repo.has_entry_between(date_range.start, date_range.end) def format_summary(self, entry: WorkEntry) -> str: tag_part = entry.tags[0].name if entry.tags else "未分组" date_part = entry.created_at.strftime("%Y-%m-%d") return f"{date_part}{self.settings.summary_separator}{tag_part}" class TagService: def __init__(self, session: Session): self.repository = TagRepository(session) def list_tags(self) -> list[Tag]: return self.repository.list_tags() def create_tag(self, name: str, description: str | None = None) -> Tag: return self.repository.create_tag(name=name, description=description)