769 lines
27 KiB
Python
769 lines
27 KiB
Python
"""End-to-end analysis pipeline: extract → filter → summarize → weekly report."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import re
|
||
import time
|
||
from collections import defaultdict
|
||
from datetime import date, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from config import LLM_BODY_CHARS, LLM_MAX_MAILS, LLM_MODEL, UPLOAD_DIR
|
||
from job_repository import (
|
||
get_job,
|
||
insert_mails,
|
||
list_mails_for_job,
|
||
purge_mail_bodies,
|
||
save_weekly_report,
|
||
update_job,
|
||
update_mail_summary,
|
||
)
|
||
from llm_client import LlmError, chat_json
|
||
from mail_parser import (
|
||
ParsedMail,
|
||
find_zip,
|
||
load_mailbox_from_zip,
|
||
)
|
||
|
||
logger = logging.getLogger("mail_summary")
|
||
|
||
THREAD_SYSTEM = """당신은 사내 메일 업무 분석기다.
|
||
제목이 아니라 본문(Body)을 기준으로 JSON만 출력한다.
|
||
외부 지식·제목만으로 추측하지 않는다. 본문에 없는 일을 만들지 않는다.
|
||
|
||
customer는 은행/고객/거래처 단위 이름이다. 예: 경남은행, 신한은행, 농협, IBK기업은행, 한화손보, 우체국.
|
||
사내 공지·인사·회계·연구기획 안내는 customer를 "사내"로 둔다.
|
||
광고·가입·프로모션 메일은 work_relevant=false.
|
||
|
||
메일 종류(Kind):
|
||
- reply(Re/답장): 이번 주 대화·업무로 본다. 본문 신규 내용만 요약.
|
||
- forward(Fw/전달): 과거 메일을 넘긴 것이다.
|
||
* 전달된 원본에 적힌 옛 업무(수년 전 견적/구축 등)를 금주 실적으로 쓰지 말 것.
|
||
* 이번에 한 일이 보이면 그것만. 예: "계약서 법무 검토 요청", "자료 전달".
|
||
* 단순 전달·참고용이고 신규 행동이 없으면 work_relevant=false.
|
||
- normal: 본문 기준 일반 요약.
|
||
|
||
스키마:
|
||
{
|
||
"customer": "경남은행",
|
||
"title": "짧은 업무 제목",
|
||
"summary": "한 문장 요약(본문 근거)",
|
||
"work_items": ["핵심 업무 1줄 (최대 2개, 각 40자 이내)"],
|
||
"work_relevant": true,
|
||
"mail_kind": "normal|reply|forward",
|
||
"horizon": "this_week|next_week|both|unknown",
|
||
"deadlines": ["날짜 메모"],
|
||
"people": ["관련자"]
|
||
}
|
||
|
||
horizon 기준:
|
||
- this_week: 금주 기간에 수행/완료/진행된 일
|
||
- next_week: 차주 예정·계획·배포 예정
|
||
- both: 금주 실적과 차주 계획이 함께 있음
|
||
- unknown: 판단 불가
|
||
"""
|
||
|
||
WEEKLY_SYSTEM = """당신은 사내 주간 업무보고 작성기다.
|
||
엑셀 주간보고에 붙일 수 있게, 짧고 요약된 JSON만 출력한다.
|
||
사람 이름 칸은 만들지 않는다. 금주/차주만 만든다.
|
||
|
||
작성 규칙:
|
||
1) 그룹 단위는 고객/은행명이다. 예: 경남은행, 농협, IBK기업은행. 사내 공지는 "사내".
|
||
2) items는 그 고객에서 이번 주(또는 차주)에 실제로 한 일/할 일만. 제목·옛 FW 원본으로 추측 금지.
|
||
3) FW/전달 요약이 "과거 견적 제출·구축"처럼 보이면 제외한다. 이번에 "검토 요청/자료 전달"한 경우만 그 행동으로 짧게 남긴다.
|
||
4) Re/답장은 본문 기준 정상 포함.
|
||
5) 불릿 1개=한 줄(되도록 40자), 고객당 최대 3개. 중복 제거.
|
||
6) 광고·뉴스레터 제외. 장황한 규정 문구 제외.
|
||
7) 금주 → this_week, 차주 → next_week.
|
||
|
||
좋은 예:
|
||
{"customer":"농협","items":["상호 통합UI WAS 분석 자료 송부 (7/16)"]}
|
||
{"customer":"동아피엠","items":["물품공급 계약서 법무 검토 요청"]}
|
||
|
||
나쁜 예:
|
||
- FW로 넘긴 몇 년 전 견적/구축을 금주 실적으로 적기
|
||
- 제목만 보고 "견적서 제출"이라고 쓰기
|
||
|
||
스키마:
|
||
{
|
||
"this_week": [
|
||
{"customer": "경남은행", "items": ["후면 인자 개발 산출물 작성", "산막공단·초장동 설치 (7/14)"]}
|
||
],
|
||
"next_week": [
|
||
{"customer": "IBK기업은행", "items": ["녹취 동의 가이드 영상 적용 예정 (7/24)"]}
|
||
]
|
||
}
|
||
"""
|
||
|
||
|
||
def scope_note(mailbox_scope: str, inbox_count: int, sent_count: int) -> str:
|
||
if mailbox_scope == "inbox" or (inbox_count > 0 and sent_count == 0):
|
||
return (
|
||
"받은 메일함만 분석되었습니다. 보낸 메일이 포함되지 않아 "
|
||
"업무 처리 여부와 전체 대화 흐름이 일부 누락될 수 있습니다."
|
||
)
|
||
if mailbox_scope == "sent" or (sent_count > 0 and inbox_count == 0):
|
||
return (
|
||
"보낸 메일함만 분석되었습니다. 최초 요청 내용과 상대방의 회신이 "
|
||
"포함되지 않아 일부 업무 배경이 누락될 수 있습니다."
|
||
)
|
||
return "받은 메일함과 보낸 메일함을 함께 분석했습니다."
|
||
|
||
|
||
def _trim(text: str, limit: int = LLM_BODY_CHARS) -> str:
|
||
text = (text or "").strip()
|
||
if len(text) <= limit:
|
||
return text
|
||
return text[: limit - 20] + "\n...[truncated]"
|
||
|
||
|
||
# Promotional / SaaS noise — never belongs in a weekly work report.
|
||
_NOISE_PATTERNS = (
|
||
re.compile(r"sign\s*up\s*for\s*cursor", re.I),
|
||
re.compile(r"you've been invited to cursor", re.I),
|
||
re.compile(r"connect your repos to cursor", re.I),
|
||
re.compile(r"canceled subscription", re.I),
|
||
re.compile(r"\bcursor\.com\b", re.I),
|
||
re.compile(r"\bcursor\.sh\b", re.I),
|
||
re.compile(r"@cursor\.", re.I),
|
||
re.compile(r"직장에서\s*copilot", re.I),
|
||
re.compile(r"copilot을\s*더\s*잘", re.I),
|
||
re.compile(r"newsletter", re.I),
|
||
re.compile(r"verify your email", re.I),
|
||
)
|
||
|
||
|
||
_FORWARD_SUBJECT_RE = re.compile(
|
||
r"^\s*((fw|fwd|전달)\s*[:：]|\[?\s*(fw|fwd|전달)\s*\]?\s*:)",
|
||
re.IGNORECASE,
|
||
)
|
||
_REPLY_SUBJECT_RE = re.compile(
|
||
r"^\s*((re|답장|회신)\s*[:：]|\[?\s*(re|답장|회신)\s*\]?\s*:)",
|
||
re.IGNORECASE,
|
||
)
|
||
# Subject alone suggests a real this-week action (keep FW for LLM).
|
||
_FORWARD_ACTION_RE = re.compile(
|
||
r"(검토\s*요청|회신\s*요청|확인\s*요청|승인\s*요청|협조\s*요청)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def _mail_kind(subject: str) -> str:
|
||
text = subject or ""
|
||
if _FORWARD_SUBJECT_RE.match(text):
|
||
return "forward"
|
||
if _REPLY_SUBJECT_RE.match(text):
|
||
return "reply"
|
||
return "normal"
|
||
|
||
|
||
def _is_noise_mail(mail: ParsedMail) -> bool:
|
||
blob = f"{mail.subject or ''} {mail.from_addr or ''}"
|
||
return any(pat.search(blob) for pat in _NOISE_PATTERNS)
|
||
|
||
|
||
def _forward_new_body_len(body: str) -> int:
|
||
"""Length of newly written text only (signature / quoted original stripped)."""
|
||
text = (body or "").replace("\u200b", "").replace(" ", " ")
|
||
text = text.replace("\xa0", " ")
|
||
lower = text.lower()
|
||
cut = len(text)
|
||
for marker in (
|
||
"보내는사람",
|
||
"보내는 사람",
|
||
"보낸 사람",
|
||
"-----original",
|
||
"begin forwarded",
|
||
"-----원본",
|
||
"from:",
|
||
):
|
||
idx = lower.find(marker)
|
||
if idx != -1:
|
||
cut = min(cut, idx)
|
||
text = text[:cut]
|
||
lines: list[str] = []
|
||
for line in text.splitlines():
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
continue
|
||
if re.match(r"^customer centric agility$", stripped, re.I):
|
||
break
|
||
if re.match(r"^-{5,}", stripped):
|
||
break
|
||
if re.match(r"^(tel|fax|mobile|e-?mail)\s*:", stripped, re.I):
|
||
break
|
||
# Greeting / name-only lines are not weekly work.
|
||
if re.fullmatch(r"안녕하세요[.]?", stripped):
|
||
continue
|
||
if re.fullmatch(r".*(연구원|책임|수석|팀장)입니다[.]?", stripped):
|
||
continue
|
||
lines.append(stripped)
|
||
return len(" ".join(lines))
|
||
|
||
|
||
def _is_forward_passthrough(mail: ParsedMail) -> bool:
|
||
"""FW/전달 with almost no new body — old mail shared as-is, not this week's work."""
|
||
if _mail_kind(mail.subject) != "forward":
|
||
return False
|
||
# Keep only when subject itself is a clear this-week request action.
|
||
if _FORWARD_ACTION_RE.search(mail.subject or ""):
|
||
return False
|
||
return _forward_new_body_len(mail.body_text or "") < 40
|
||
|
||
|
||
def _is_noise_thread(thread: list[ParsedMail]) -> bool:
|
||
if any(_is_noise_mail(mail) for mail in thread):
|
||
return True
|
||
# Drop threads that are only empty/short forwards of old mail.
|
||
return bool(thread) and all(_is_forward_passthrough(mail) for mail in thread)
|
||
|
||
|
||
def _is_noise_summary(item: dict[str, Any]) -> bool:
|
||
if item.get("work_relevant") is False:
|
||
return True
|
||
blob = " ".join(
|
||
str(item.get(key) or "")
|
||
for key in ("customer", "title", "summary")
|
||
)
|
||
items = item.get("work_items") or item.get("items") or []
|
||
if isinstance(items, list):
|
||
blob = f"{blob} {' '.join(str(x) for x in items)}"
|
||
return any(pat.search(blob) for pat in _NOISE_PATTERNS)
|
||
|
||
|
||
def _mail_blob(mail: ParsedMail) -> str:
|
||
mailed = mail.mailed_at.isoformat() if mail.mailed_at else ""
|
||
attachments = ", ".join(mail.attachment_names) if mail.attachment_names else "(없음)"
|
||
kind = _mail_kind(mail.subject)
|
||
body = _trim(mail.body_text)
|
||
kind_note = {
|
||
"forward": (
|
||
"전달(FW) 메일이다. Body는 이번에 새로 쓴 부분만이다. "
|
||
"원본에 있던 과거 견적/구축 등을 금주 실적으로 쓰지 말고, "
|
||
"신규 행동(검토 요청·자료 전달 등)만 요약하라. "
|
||
"신규 행동이 없으면 work_relevant=false."
|
||
),
|
||
"reply": "답장(Re) 메일이다. Body 신규 대화 내용만 요약하라.",
|
||
"normal": "일반 메일이다. Body를 반드시 읽고 요약하라. 제목만으로 업무를 만들지 마라.",
|
||
}[kind]
|
||
if not body.strip():
|
||
body = "(신규 본문 없음 — 제목만으로 업무를 추측하지 말 것)"
|
||
return (
|
||
f"[{mail.mailbox}] {mailed}\n"
|
||
f"Kind: {kind}\n"
|
||
f"Note: {kind_note}\n"
|
||
f"From: {mail.from_addr}\n"
|
||
f"To: {mail.to_addrs}\n"
|
||
f"Subject: {mail.subject}\n"
|
||
f"Attachments: {attachments}\n"
|
||
f"Body:\n{body}\n"
|
||
)
|
||
|
||
|
||
def _heuristic_thread_summary(thread: list[ParsedMail]) -> dict[str, Any]:
|
||
latest = thread[-1]
|
||
people = sorted(
|
||
{
|
||
p.strip()
|
||
for mail in thread
|
||
for p in (mail.from_addr, *mail.to_addrs.split(","))
|
||
if p and p.strip()
|
||
}
|
||
)
|
||
title = latest.subject or thread[0].thread_key
|
||
summary = _trim(latest.body_text, 280) or "(본문 없음)"
|
||
return {
|
||
"customer": "기타",
|
||
"title": title,
|
||
"summary": summary,
|
||
"work_items": [title] if title else [],
|
||
"work_relevant": True,
|
||
"horizon": "unknown",
|
||
"deadlines": [],
|
||
"people": people[:8],
|
||
"heuristic": True,
|
||
}
|
||
|
||
|
||
_MAX_ITEMS_PER_CUSTOMER = 3
|
||
_MAX_ITEM_CHARS = 48
|
||
|
||
|
||
def _compact_item(text: str) -> str:
|
||
text = " ".join(str(text).split())
|
||
if len(text) <= _MAX_ITEM_CHARS:
|
||
return text
|
||
return text[: _MAX_ITEM_CHARS - 1].rstrip() + "…"
|
||
|
||
|
||
def _normalize_side(rows: Any) -> list[dict[str, Any]]:
|
||
if not isinstance(rows, list):
|
||
return []
|
||
merged: dict[str, list[str]] = {}
|
||
order: list[str] = []
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
customer = str(row.get("customer") or row.get("project") or "기타").strip() or "기타"
|
||
raw_items = row.get("items") or row.get("bullets") or row.get("work_items") or []
|
||
if isinstance(raw_items, str):
|
||
raw_items = [raw_items]
|
||
if not isinstance(raw_items, list):
|
||
raw_items = []
|
||
items = [str(x).strip() for x in raw_items if str(x).strip()]
|
||
# Legacy fallbacks from older schema
|
||
if not items:
|
||
detail = row.get("detail") or row.get("summary") or row.get("title")
|
||
if detail:
|
||
items = [str(detail).strip()]
|
||
if customer not in merged:
|
||
merged[customer] = []
|
||
order.append(customer)
|
||
for item in items:
|
||
compact = _compact_item(item)
|
||
if compact and compact not in merged[customer]:
|
||
merged[customer].append(compact)
|
||
if len(merged[customer]) >= _MAX_ITEMS_PER_CUSTOMER:
|
||
break
|
||
return [
|
||
{"customer": name, "items": merged[name][:_MAX_ITEMS_PER_CUSTOMER]}
|
||
for name in order
|
||
if merged[name]
|
||
]
|
||
|
||
|
||
def normalize_weekly_report(report: dict[str, Any]) -> dict[str, Any]:
|
||
this_week = [
|
||
row for row in _normalize_side(report.get("this_week")) if not _is_noise_summary(row)
|
||
]
|
||
next_week = [
|
||
row for row in _normalize_side(report.get("next_week")) if not _is_noise_summary(row)
|
||
]
|
||
return {
|
||
"this_week": this_week,
|
||
"next_week": next_week,
|
||
"heuristic": bool(report.get("heuristic")),
|
||
}
|
||
|
||
|
||
def _heuristic_weekly(thread_summaries: list[dict[str, Any]]) -> dict[str, Any]:
|
||
this_week: list[dict[str, Any]] = []
|
||
next_week: list[dict[str, Any]] = []
|
||
for item in thread_summaries:
|
||
if _is_noise_summary(item):
|
||
continue
|
||
customer = str(item.get("customer") or "기타").strip() or "기타"
|
||
work_items = item.get("work_items") or []
|
||
if not isinstance(work_items, list) or not work_items:
|
||
title = item.get("title") or item.get("summary")
|
||
work_items = [title] if title else []
|
||
work_items = [str(x).strip() for x in work_items if str(x).strip()][:2]
|
||
if not work_items:
|
||
continue
|
||
row = {"customer": customer, "items": work_items}
|
||
horizon = str(item.get("horizon") or "unknown")
|
||
if horizon in ("next_week",):
|
||
next_week.append(row)
|
||
elif horizon == "both":
|
||
this_week.append(row)
|
||
next_week.append(row)
|
||
else:
|
||
this_week.append(row)
|
||
return normalize_weekly_report(
|
||
{"this_week": this_week, "next_week": next_week, "heuristic": True}
|
||
)
|
||
|
||
|
||
def _format_side_markdown(rows: list[dict[str, Any]]) -> list[str]:
|
||
if not rows:
|
||
return ["(해당 항목 없음)", ""]
|
||
lines: list[str] = []
|
||
for row in rows:
|
||
customer = row.get("customer") or "기타"
|
||
lines.append(str(customer))
|
||
items = row.get("items") or []
|
||
if not items:
|
||
lines.append("- (세부 항목 없음)")
|
||
else:
|
||
for item in items:
|
||
lines.append(f"- {item}")
|
||
lines.append("")
|
||
return lines
|
||
|
||
|
||
def report_to_markdown(
|
||
report: dict[str, Any],
|
||
*,
|
||
note: str,
|
||
inbox: int,
|
||
sent: int,
|
||
date_from: str,
|
||
date_to: str,
|
||
next_from: str,
|
||
next_to: str,
|
||
) -> str:
|
||
_ = (note, inbox, sent)
|
||
report = normalize_weekly_report(report)
|
||
lines = [
|
||
f"금주 ({date_from} ~ {date_to})",
|
||
"",
|
||
*_format_side_markdown(report["this_week"]),
|
||
f"차주 ({next_from} ~ {next_to})",
|
||
"",
|
||
*_format_side_markdown(report["next_week"]),
|
||
]
|
||
return "\n".join(lines).strip() + "\n"
|
||
|
||
|
||
async def _summarize_thread(
|
||
thread_key: str,
|
||
thread: list[ParsedMail],
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
user = (
|
||
f"스레드 키: {thread_key}\n"
|
||
f"메일 수: {len(thread)}\n"
|
||
"지시: 각 메일의 Body를 읽고 요약하라. Subject만으로 업무를 만들지 마라. "
|
||
"FW/전달의 과거 원본 업무를 금주 실적으로 쓰지 마라.\n\n"
|
||
+ "\n---\n".join(_mail_blob(m) for m in thread)
|
||
)
|
||
try:
|
||
data = await chat_json(system=THREAD_SYSTEM, user=user, model=model)
|
||
data["heuristic"] = False
|
||
return data
|
||
except LlmError:
|
||
logger.warning("Thread LLM summary failed; using heuristic (%s)", thread_key)
|
||
return _heuristic_thread_summary(thread)
|
||
|
||
|
||
async def _summarize_week(
|
||
*,
|
||
date_from: str,
|
||
date_to: str,
|
||
next_from: str,
|
||
next_to: str,
|
||
thread_summaries: list[dict[str, Any]],
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
compact = []
|
||
for item in thread_summaries:
|
||
if _is_noise_summary(item):
|
||
continue
|
||
compact.append(
|
||
{
|
||
"customer": item.get("customer"),
|
||
"title": item.get("title"),
|
||
"summary": item.get("summary"),
|
||
"work_items": item.get("work_items"),
|
||
"mail_kind": item.get("mail_kind"),
|
||
"horizon": item.get("horizon"),
|
||
"deadlines": item.get("deadlines"),
|
||
}
|
||
)
|
||
user = (
|
||
f"금주 기간: {date_from} ~ {date_to}\n"
|
||
f"차주 기간: {next_from} ~ {next_to}\n"
|
||
f"스레드 수: {len(compact)}\n"
|
||
f"스레드 요약 JSON:\n{compact}\n\n"
|
||
"고객/은행 단위로 금주·차주 JSON을 작성하라. "
|
||
"짧게 요약하고, 고객당 불릿 최대 3개, 광고성 메일은 제외하라."
|
||
)
|
||
try:
|
||
data = await chat_json(system=WEEKLY_SYSTEM, user=user, model=model)
|
||
data["heuristic"] = False
|
||
return normalize_weekly_report(data)
|
||
except LlmError:
|
||
logger.warning("Weekly LLM summary failed; using heuristic")
|
||
return _heuristic_weekly(thread_summaries)
|
||
|
||
|
||
async def _set_progress(
|
||
job_id: int,
|
||
*,
|
||
status: str | None = None,
|
||
pct: int,
|
||
label: str,
|
||
detail: str = "",
|
||
inbox_count: int | None = None,
|
||
sent_count: int | None = None,
|
||
filtered_inbox_count: int | None = None,
|
||
filtered_sent_count: int | None = None,
|
||
started: bool = False,
|
||
completed: bool = False,
|
||
error_message: str | None = None,
|
||
) -> None:
|
||
await update_job(
|
||
job_id,
|
||
status=status,
|
||
progress_pct=max(0, min(100, pct)),
|
||
progress_label=label,
|
||
progress_detail=detail,
|
||
inbox_count=inbox_count,
|
||
sent_count=sent_count,
|
||
filtered_inbox_count=filtered_inbox_count,
|
||
filtered_sent_count=filtered_sent_count,
|
||
started=started,
|
||
completed=completed,
|
||
error_message=error_message,
|
||
)
|
||
|
||
|
||
async def run_analysis_job(job_id: int) -> None:
|
||
job = await get_job(job_id)
|
||
if not job:
|
||
logger.error("Job %s not found", job_id)
|
||
return
|
||
|
||
job_dir = UPLOAD_DIR / str(job_id)
|
||
model = job.get("model") or LLM_MODEL
|
||
date_from = date.fromisoformat(job["date_from"])
|
||
date_to = date.fromisoformat(job["date_to"])
|
||
scope = job["mailbox_scope"]
|
||
loop = asyncio.get_running_loop()
|
||
|
||
try:
|
||
await _set_progress(
|
||
job_id,
|
||
status="extracting",
|
||
pct=3,
|
||
label="분석 준비 중",
|
||
detail="업로드된 zip을 확인하고 있습니다.",
|
||
started=True,
|
||
)
|
||
|
||
def _report(pct: int, label: str, detail: str, **counts: int) -> None:
|
||
asyncio.run_coroutine_threadsafe(
|
||
_set_progress(
|
||
job_id,
|
||
status="filtering",
|
||
pct=pct,
|
||
label=label,
|
||
detail=detail,
|
||
**counts,
|
||
),
|
||
loop,
|
||
)
|
||
|
||
def _load() -> tuple[list[ParsedMail], list[ParsedMail], int, int]:
|
||
inbox: list[ParsedMail] = []
|
||
sent: list[ParsedMail] = []
|
||
inbox_total = 0
|
||
sent_total = 0
|
||
|
||
do_inbox = scope in ("inbox", "both")
|
||
do_sent = scope in ("sent", "both")
|
||
|
||
if do_inbox:
|
||
z = find_zip(job_dir, "inbox")
|
||
if z:
|
||
last_push = [0.0]
|
||
|
||
def on_inbox(scanned: int, total: int, matched: int) -> None:
|
||
now = time.monotonic()
|
||
if scanned != total and now - last_push[0] < 1.2:
|
||
return
|
||
last_push[0] = now
|
||
ratio = (scanned / total) if total else 1
|
||
pct = 8 + int(ratio * 37)
|
||
_report(
|
||
pct,
|
||
"받은 메일함 스캔 중",
|
||
f"대용량 zip에서 메일을 읽는 중입니다. 검사 {scanned:,}/{total:,} · 기간 매칭 {matched:,}",
|
||
inbox_count=total,
|
||
filtered_inbox_count=matched,
|
||
)
|
||
|
||
_report(8, "받은 메일함 스캔 중", "받은 메일 zip 목록을 준비하는 중입니다.")
|
||
inbox, inbox_total = load_mailbox_from_zip(
|
||
z,
|
||
"inbox",
|
||
date_from=date_from,
|
||
date_to=date_to,
|
||
on_progress=on_inbox,
|
||
)
|
||
|
||
if do_sent:
|
||
z = find_zip(job_dir, "sent")
|
||
if z:
|
||
last_push = [0.0]
|
||
|
||
def on_sent(scanned: int, total: int, matched: int) -> None:
|
||
now = time.monotonic()
|
||
if scanned != total and now - last_push[0] < 1.2:
|
||
return
|
||
last_push[0] = now
|
||
ratio = (scanned / total) if total else 1
|
||
pct = 48 + int(ratio * 32)
|
||
_report(
|
||
pct,
|
||
"보낸 메일함 스캔 중",
|
||
f"대용량 zip에서 메일을 읽는 중입니다. 검사 {scanned:,}/{total:,} · 기간 매칭 {matched:,}",
|
||
sent_count=total,
|
||
filtered_sent_count=matched,
|
||
inbox_count=inbox_total,
|
||
filtered_inbox_count=len(inbox),
|
||
)
|
||
|
||
_report(
|
||
48,
|
||
"보낸 메일함 스캔 중",
|
||
"보낸 메일 zip 목록을 준비하는 중입니다.",
|
||
inbox_count=inbox_total,
|
||
filtered_inbox_count=len(inbox),
|
||
)
|
||
sent, sent_total = load_mailbox_from_zip(
|
||
z,
|
||
"sent",
|
||
date_from=date_from,
|
||
date_to=date_to,
|
||
on_progress=on_sent,
|
||
)
|
||
|
||
return inbox, sent, inbox_total, sent_total
|
||
|
||
await _set_progress(
|
||
job_id,
|
||
status="filtering",
|
||
pct=6,
|
||
label="메일 기간 필터링 시작",
|
||
detail=f"{job['date_from']} ~ {job['date_to']} 구간의 메일만 추립니다. 수 GB zip은 수 분 걸릴 수 있습니다.",
|
||
)
|
||
inbox, sent, inbox_total, sent_total = await asyncio.to_thread(_load)
|
||
inbox = [
|
||
m for m in inbox if not _is_noise_mail(m) and not _is_forward_passthrough(m)
|
||
]
|
||
sent = [
|
||
m for m in sent if not _is_noise_mail(m) and not _is_forward_passthrough(m)
|
||
]
|
||
selected = inbox + sent
|
||
if len(selected) > LLM_MAX_MAILS:
|
||
selected = selected[:LLM_MAX_MAILS]
|
||
|
||
await _set_progress(
|
||
job_id,
|
||
status="analyzing",
|
||
pct=82,
|
||
label="분석용 메일 저장 중",
|
||
detail=f"기간 내 메일 {len(selected):,}건을 저장합니다. (전체 검사 받은 {inbox_total:,} / 보낸 {sent_total:,})",
|
||
inbox_count=inbox_total,
|
||
sent_count=sent_total,
|
||
filtered_inbox_count=len(inbox),
|
||
filtered_sent_count=len(sent),
|
||
)
|
||
|
||
await insert_mails(job_id, selected)
|
||
stored = await list_mails_for_job(job_id)
|
||
|
||
by_thread: dict[str, list[ParsedMail]] = defaultdict(list)
|
||
id_by_message: dict[tuple[str, str], int] = {}
|
||
original_by_key = {(m.mailbox, m.message_id): m for m in selected}
|
||
for row in stored:
|
||
key = (row["mailbox"], row["message_id"] or "")
|
||
original = original_by_key.get(key)
|
||
if not original:
|
||
continue
|
||
id_by_message[key] = row["id"]
|
||
by_thread[original.thread_key].append(original)
|
||
|
||
thread_items = [
|
||
(key, thread)
|
||
for key, thread in by_thread.items()
|
||
if not _is_noise_thread(thread)
|
||
]
|
||
skipped_noise = len(by_thread) - len(thread_items)
|
||
if skipped_noise:
|
||
logger.info("Job %s skipped %s noise thread(s)", job_id, skipped_noise)
|
||
thread_summaries: list[dict[str, Any]] = []
|
||
total_threads = max(len(thread_items), 1)
|
||
for idx, (thread_key, thread) in enumerate(thread_items, start=1):
|
||
pct = 84 + int((idx / total_threads) * 10)
|
||
await _set_progress(
|
||
job_id,
|
||
status="analyzing",
|
||
pct=pct,
|
||
label="로컬 LLM 스레드 요약 중",
|
||
detail=f"업무 스레드 {idx}/{len(thread_items)} 분석 중 · {thread_key[:60]}",
|
||
inbox_count=inbox_total,
|
||
sent_count=sent_total,
|
||
filtered_inbox_count=len(inbox),
|
||
filtered_sent_count=len(sent),
|
||
)
|
||
summary = await _summarize_thread(thread_key, thread, model)
|
||
if _is_noise_summary(summary):
|
||
continue
|
||
thread_summaries.append(summary)
|
||
for mail in thread:
|
||
mail_id = id_by_message.get((mail.mailbox, mail.message_id))
|
||
if mail_id:
|
||
await update_mail_summary(mail_id, summary)
|
||
|
||
await _set_progress(
|
||
job_id,
|
||
status="analyzing",
|
||
pct=96,
|
||
label="주간 업무보고 작성 중",
|
||
detail="스레드 요약을 모아 주간 보고 형식으로 정리하고 있습니다.",
|
||
inbox_count=inbox_total,
|
||
sent_count=sent_total,
|
||
filtered_inbox_count=len(inbox),
|
||
filtered_sent_count=len(sent),
|
||
)
|
||
next_from_d = date_to + timedelta(days=1)
|
||
next_to_d = date_to + timedelta(days=7)
|
||
next_from = next_from_d.isoformat()
|
||
next_to = next_to_d.isoformat()
|
||
weekly = await _summarize_week(
|
||
date_from=job["date_from"],
|
||
date_to=job["date_to"],
|
||
next_from=next_from,
|
||
next_to=next_to,
|
||
thread_summaries=thread_summaries,
|
||
model=model,
|
||
)
|
||
note = scope_note(scope, len(inbox), len(sent))
|
||
markdown = report_to_markdown(
|
||
weekly,
|
||
note=note,
|
||
inbox=len(inbox),
|
||
sent=len(sent),
|
||
date_from=job["date_from"],
|
||
date_to=job["date_to"],
|
||
next_from=next_from,
|
||
next_to=next_to,
|
||
)
|
||
await save_weekly_report(
|
||
job_id=job_id,
|
||
model=model,
|
||
scope_note=note,
|
||
report=weekly,
|
||
report_markdown=markdown,
|
||
)
|
||
await purge_mail_bodies(job_id)
|
||
await _set_progress(
|
||
job_id,
|
||
status="completed",
|
||
pct=100,
|
||
label="분석 완료",
|
||
detail=f"주간 보고 생성 완료 · 스레드 {len(thread_summaries)}개",
|
||
inbox_count=inbox_total,
|
||
sent_count=sent_total,
|
||
filtered_inbox_count=len(inbox),
|
||
filtered_sent_count=len(sent),
|
||
completed=True,
|
||
)
|
||
logger.info("Job %s completed (%s threads)", job_id, len(thread_summaries))
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.exception("Job %s failed", job_id)
|
||
await _set_progress(
|
||
job_id,
|
||
status="failed",
|
||
pct=100,
|
||
label="분석 실패",
|
||
detail=str(exc)[:300],
|
||
error_message=str(exc)[:1000],
|
||
completed=True,
|
||
)
|