Frist commit

backend/.env

@@ -0,0 +1,13 @@
+# DATABASE_URL is injected by compose (points at the mail_summary postgres service).
+
+CORS_ORIGINS=https://ms.takits.me,http://localhost:4888,http://localhost:5173
+
+# Local LLM (Ollama on Mac).
+LLM_BASE_URL=https://mac-ollama.takits.me
+LLM_MODEL=gemma4:31b-mlx
+LLM_TIMEOUT_SECONDS=300
+LLM_MAX_MAILS=80
+LLM_BODY_CHARS=3000
+
+UPLOAD_DIR=/app/uploads
+INCOMING_DIR=/app/incoming

backend/Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# tzdata lets the TZ env var (Asia/Seoul) resolve so log timestamps follow KST.
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends tzdata \
+    && rm -rf /var/lib/apt/lists/*
+ENV TZ=Asia/Seoul
+
+COPY backend/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY backend/ .
+
+RUN mkdir -p /app/uploads /app/logs
+
+ENV PYTHONUNBUFFERED=1
+
+EXPOSE 8000
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

backend/app.py

@@ -0,0 +1,308 @@
+"""mail_summary API — local LLM weekly mail work summary."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import shutil
+import uuid
+from contextlib import asynccontextmanager
+from datetime import date
+from pathlib import Path
+from typing import Literal
+
+from fastapi import FastAPI, File, Form, HTTPException, UploadFile
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+
+from config import CORS_ORIGINS, INCOMING_DIR, LLM_MODEL, UPLOAD_DIR
+from db.pool import close_pool, init_pool, pool_enabled
+from job_repository import create_job, ensure_tables, get_job, get_report, list_jobs
+from llm_client import ping_llm
+from pipeline import run_analysis_job
+
+logger = logging.getLogger("mail_summary")
+logging.basicConfig(level=logging.INFO)
+logger.setLevel(logging.INFO)
+
+MailboxScope = Literal["inbox", "sent", "both"]
+Mailbox = Literal["inbox", "sent"]
+STAGING_DIR = UPLOAD_DIR / "staging"
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
+    STAGING_DIR.mkdir(parents=True, exist_ok=True)
+    INCOMING_DIR.mkdir(parents=True, exist_ok=True)
+    await init_pool()
+    await ensure_tables()
+    yield
+    await close_pool()
+
+
+app = FastAPI(title="Mail Summary API", version="1.0.0", lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=CORS_ORIGINS or ["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+class JobCreateResponse(BaseModel):
+    job: dict
+
+
+class HealthResponse(BaseModel):
+    status: str
+    database: str
+    model: str = Field(default="")
+    llm: dict = Field(default_factory=dict)
+
+
+class UploadResponse(BaseModel):
+    upload_id: str
+    mailbox: str
+    filename: str
+    size_bytes: int
+    source: str  # upload | incoming
+
+
+def _staging_path(upload_id: str) -> Path:
+    path = (STAGING_DIR / upload_id).resolve()
+    if not str(path).startswith(str(STAGING_DIR.resolve())):
+        raise HTTPException(status_code=400, detail="invalid_upload_id")
+    return path
+
+
+def _write_staging_meta(stage_dir: Path, meta: dict) -> None:
+    (stage_dir / "meta.json").write_text(json.dumps(meta, ensure_ascii=False), encoding="utf-8")
+
+
+def _read_staging_meta(upload_id: str) -> dict:
+    stage_dir = _staging_path(upload_id)
+    meta_path = stage_dir / "meta.json"
+    if not meta_path.is_file():
+        raise HTTPException(status_code=404, detail="upload_not_found")
+    return json.loads(meta_path.read_text(encoding="utf-8"))
+
+
+def _attach_source(src: Path, stage_dir: Path, mailbox: str, filename: str) -> Path:
+    dest = stage_dir / f"{mailbox}_{filename}"
+    if dest.exists() or dest.is_symlink():
+        dest.unlink()
+    try:
+        dest.symlink_to(src)
+    except OSError:
+        shutil.copy2(src, dest)
+    return dest
+
+
+def _find_incoming_match(filename: str, size_bytes: int) -> Path | None:
+    candidate = INCOMING_DIR / Path(filename).name
+    if not candidate.is_file():
+        return None
+    try:
+        if candidate.stat().st_size != size_bytes:
+            return None
+    except OSError:
+        return None
+    return candidate.resolve()
+
+
+def _attach_staging_to_job(upload_id: str, job_dir: Path, mailbox: str) -> str:
+    """Copy/link staged zip into a job dir, keeping staging for re-analysis."""
+    meta = _read_staging_meta(upload_id)
+    if meta.get("mailbox") != mailbox:
+        raise HTTPException(status_code=400, detail="upload_mailbox_mismatch")
+    stage_dir = _staging_path(upload_id)
+    filename = meta["filename"]
+    src = stage_dir / f"{mailbox}_{filename}"
+    if not src.exists() and not src.is_symlink():
+        raise HTTPException(status_code=404, detail="upload_file_missing")
+    dest = job_dir / f"{mailbox}_{filename}"
+    if dest.exists() or dest.is_symlink():
+        dest.unlink()
+    # Prefer symlink so multi-GB files are not duplicated.
+    try:
+        target = src.resolve() if src.is_symlink() else src
+        dest.symlink_to(target)
+    except OSError:
+        shutil.copy2(src, dest)
+    return filename
+
+
+@app.get("/api/health", response_model=HealthResponse)
+async def health() -> HealthResponse:
+    llm = await ping_llm()
+    return HealthResponse(
+        status="ok",
+        database="up" if pool_enabled() else "down",
+        model=LLM_MODEL,
+        llm=llm,
+    )
+
+
+@app.post("/api/uploads/resolve", response_model=UploadResponse)
+async def api_resolve_upload(
+    mailbox: Mailbox = Form(...),
+    filename: str = Form(...),
+    size_bytes: int = Form(...),
+) -> UploadResponse:
+    """If the selected file already exists on the incoming mount, stage it without re-upload."""
+    match = _find_incoming_match(filename, size_bytes)
+    if not match:
+        raise HTTPException(status_code=404, detail="incoming_match_not_found")
+
+    upload_id = uuid.uuid4().hex
+    stage_dir = _staging_path(upload_id)
+    stage_dir.mkdir(parents=True, exist_ok=True)
+    safe_name = Path(filename).name
+    _attach_source(match, stage_dir, mailbox, safe_name)
+    meta = {
+        "upload_id": upload_id,
+        "mailbox": mailbox,
+        "filename": safe_name,
+        "size_bytes": size_bytes,
+        "source": "incoming",
+    }
+    _write_staging_meta(stage_dir, meta)
+    return UploadResponse(**meta)
+
+
+@app.post("/api/uploads", response_model=UploadResponse)
+async def api_upload_zip(
+    mailbox: Mailbox = Form(...),
+    file: UploadFile = File(...),
+) -> UploadResponse:
+    name = (file.filename or "").lower()
+    if not name.endswith(".zip"):
+        raise HTTPException(status_code=400, detail="file_must_be_zip")
+
+    upload_id = uuid.uuid4().hex
+    stage_dir = _staging_path(upload_id)
+    stage_dir.mkdir(parents=True, exist_ok=True)
+    safe_name = Path(file.filename or f"{mailbox}.zip").name
+    dest = stage_dir / f"{mailbox}_{safe_name}"
+    size = 0
+    try:
+        with dest.open("wb") as out:
+            while True:
+                chunk = await file.read(1024 * 1024)
+                if not chunk:
+                    break
+                out.write(chunk)
+                size += len(chunk)
+    except Exception:
+        shutil.rmtree(stage_dir, ignore_errors=True)
+        logger.exception("Upload failed")
+        raise HTTPException(status_code=500, detail="upload_failed") from None
+
+    meta = {
+        "upload_id": upload_id,
+        "mailbox": mailbox,
+        "filename": safe_name,
+        "size_bytes": size,
+        "source": "upload",
+    }
+    _write_staging_meta(stage_dir, meta)
+    return UploadResponse(**meta)
+
+
+@app.delete("/api/uploads/{upload_id}")
+async def api_delete_upload(upload_id: str) -> dict:
+    stage_dir = _staging_path(upload_id)
+    if stage_dir.is_dir():
+        shutil.rmtree(stage_dir, ignore_errors=True)
+    return {"ok": True, "upload_id": upload_id}
+
+
+@app.get("/api/jobs")
+async def api_list_jobs(limit: int = 20) -> dict:
+    if not pool_enabled():
+        raise HTTPException(status_code=503, detail="database_unavailable")
+    jobs = await list_jobs(limit=max(1, min(limit, 100)))
+    return {"jobs": jobs}
+
+
+@app.get("/api/jobs/{job_id}")
+async def api_get_job(job_id: int) -> dict:
+    if not pool_enabled():
+        raise HTTPException(status_code=503, detail="database_unavailable")
+    job = await get_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="job_not_found")
+    return {"job": job}
+
+
+@app.get("/api/jobs/{job_id}/report")
+async def api_get_report(job_id: int) -> dict:
+    if not pool_enabled():
+        raise HTTPException(status_code=503, detail="database_unavailable")
+    job = await get_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="job_not_found")
+    report = await get_report(job_id)
+    return {"job": job, "report": report}
+
+
+@app.post("/api/jobs", response_model=JobCreateResponse)
+async def api_create_job(
+    date_from: date = Form(...),
+    date_to: date = Form(...),
+    mailbox_scope: MailboxScope = Form("both"),
+    model: str = Form(""),
+    inbox_upload_id: str = Form(""),
+    sent_upload_id: str = Form(""),
+) -> JobCreateResponse:
+    if not pool_enabled():
+        raise HTTPException(status_code=503, detail="database_unavailable")
+    if date_from > date_to:
+        raise HTTPException(status_code=400, detail="invalid_date_range")
+
+    inbox_upload_id = inbox_upload_id.strip()
+    sent_upload_id = sent_upload_id.strip()
+
+    if mailbox_scope in ("inbox", "both") and not inbox_upload_id:
+        raise HTTPException(status_code=400, detail="inbox_zip_required")
+    if mailbox_scope in ("sent", "both") and not sent_upload_id:
+        raise HTTPException(status_code=400, detail="sent_zip_required")
+
+    selected_model = (model or LLM_MODEL).strip() or LLM_MODEL
+    job_token = uuid.uuid4().hex
+    job_dir = UPLOAD_DIR / job_token
+    job_dir.mkdir(parents=True, exist_ok=True)
+
+    try:
+        inbox_name = (
+            _attach_staging_to_job(inbox_upload_id, job_dir, "inbox") if inbox_upload_id else None
+        )
+        sent_name = (
+            _attach_staging_to_job(sent_upload_id, job_dir, "sent") if sent_upload_id else None
+        )
+        job = await create_job(
+            mailbox_scope=mailbox_scope,
+            date_from=date_from,
+            date_to=date_to,
+            model=selected_model,
+            inbox_zip_name=inbox_name,
+            sent_zip_name=sent_name,
+        )
+        final_dir = UPLOAD_DIR / str(job["id"])
+        if final_dir.exists():
+            shutil.rmtree(final_dir)
+        job_dir.rename(final_dir)
+    except HTTPException:
+        shutil.rmtree(job_dir, ignore_errors=True)
+        raise
+    except Exception:
+        shutil.rmtree(job_dir, ignore_errors=True)
+        logger.exception("Failed to create analysis job")
+        raise HTTPException(status_code=500, detail="job_create_failed") from None
+
+    asyncio.create_task(run_analysis_job(job["id"]))
+    return JobCreateResponse(job=job)

backend/config.py

@@ -0,0 +1,39 @@
+"""mail_summary backend configuration."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+BASE_DIR = Path(__file__).resolve().parent
+
+DEFAULT_DATABASE_URL = "postgresql://mail_summary:strongpassword@localhost:5437/mail_summary"
+
+
+def _load_backend_env() -> None:
+    """Load backend/.env. Docker Compose also injects the same file via env_file."""
+    path = BASE_DIR / ".env"
+    if path.is_file():
+        load_dotenv(path)
+
+
+_load_backend_env()
+
+DATABASE_URL = os.getenv("DATABASE_URL", DEFAULT_DATABASE_URL).strip()
+
+_cors = os.getenv("CORS_ORIGINS", "http://localhost:5173,http://localhost:4888")
+CORS_ORIGINS = [origin.strip() for origin in _cors.split(",") if origin.strip()]
+
+LLM_BASE_URL = os.getenv("LLM_BASE_URL", "http://host.docker.internal:11434").rstrip("/")
+LLM_MODEL = os.getenv("LLM_MODEL", "gemma4:31b-mlx").strip()
+LLM_TIMEOUT_SECONDS = float(os.getenv("LLM_TIMEOUT_SECONDS", "300"))
+LLM_MAX_MAILS = int(os.getenv("LLM_MAX_MAILS", "80"))
+LLM_BODY_CHARS = int(os.getenv("LLM_BODY_CHARS", "3000"))
+
+UPLOAD_DIR = Path(os.getenv("UPLOAD_DIR", str(BASE_DIR / "uploads"))).resolve()
+UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
+
+INCOMING_DIR = Path(os.getenv("INCOMING_DIR", str(BASE_DIR.parent / "incoming"))).resolve()
+INCOMING_DIR.mkdir(parents=True, exist_ok=True)

backend/db/__init__.py

@@ -0,0 +1 @@
+"""Database helpers."""

backend/db/pool.py

@@ -0,0 +1,56 @@
+"""PostgreSQL connection pool."""
+
+from __future__ import annotations
+
+import logging
+
+from psycopg_pool import AsyncConnectionPool
+
+from config import DATABASE_URL
+
+logger = logging.getLogger("mail_summary")
+
+_pool: AsyncConnectionPool | None = None
+_pool_enabled = False
+
+
+async def init_pool() -> None:
+    global _pool, _pool_enabled
+    if _pool is not None:
+        return
+
+    try:
+        pool = AsyncConnectionPool(
+            conninfo=DATABASE_URL,
+            min_size=1,
+            max_size=10,
+            open=False,
+        )
+        await pool.open()
+        _pool = pool
+        _pool_enabled = True
+        logger.info("PostgreSQL connection pool initialized")
+    except Exception:
+        _pool = None
+        _pool_enabled = False
+        logger.exception("PostgreSQL unavailable")
+
+
+async def close_pool() -> None:
+    global _pool, _pool_enabled
+    if _pool is None:
+        return
+    await _pool.close()
+    _pool = None
+    _pool_enabled = False
+    logger.info("PostgreSQL connection pool closed")
+
+
+def pool_enabled() -> bool:
+    return _pool_enabled and _pool is not None
+
+
+def get_pool() -> AsyncConnectionPool:
+    if _pool is None:
+        raise RuntimeError("Database pool is not initialized")
+    return _pool

backend/job_repository.py

@@ -0,0 +1,295 @@
+"""Persistence for analysis jobs, mails, and weekly reports."""
+
+from __future__ import annotations
+
+import logging
+from datetime import date
+from typing import Any
+
+from psycopg.rows import dict_row
+from psycopg.types.json import Jsonb
+
+from db.pool import get_pool, pool_enabled
+from mail_parser import ParsedMail
+from queries import Jobs, Mails, Reports, Schema
+
+logger = logging.getLogger("mail_summary")
+
+
+def _row_to_job(row: Any) -> dict[str, Any]:
+    return {
+        "id": row["id"],
+        "status": row["status"],
+        "mailbox_scope": row["mailbox_scope"],
+        "date_from": row["date_from"].isoformat() if row["date_from"] else None,
+        "date_to": row["date_to"].isoformat() if row["date_to"] else None,
+        "model": row["model"],
+        "inbox_zip_name": row["inbox_zip_name"],
+        "sent_zip_name": row["sent_zip_name"],
+        "inbox_count": row["inbox_count"],
+        "sent_count": row["sent_count"],
+        "filtered_inbox_count": row["filtered_inbox_count"],
+        "filtered_sent_count": row["filtered_sent_count"],
+        "progress_pct": row.get("progress_pct") or 0,
+        "progress_label": row.get("progress_label") or "",
+        "progress_detail": row.get("progress_detail") or "",
+        "error_message": row["error_message"],
+        "created_at": row["created_at"].isoformat() if row["created_at"] else None,
+        "started_at": row["started_at"].isoformat() if row["started_at"] else None,
+        "completed_at": row["completed_at"].isoformat() if row["completed_at"] else None,
+    }
+
+
+def _row_to_report(row: Any) -> dict[str, Any]:
+    return {
+        "id": row["id"],
+        "job_id": row["job_id"],
+        "model": row["model"],
+        "scope_note": row["scope_note"],
+        "report": row["report"] or {},
+        "report_markdown": row["report_markdown"],
+        "created_at": row["created_at"].isoformat() if row["created_at"] else None,
+        "updated_at": row["updated_at"].isoformat() if row["updated_at"] else None,
+    }
+
+
+async def ensure_tables() -> None:
+    if not pool_enabled():
+        return
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(Schema.ENSURE_ANALYSIS_JOBS)
+            await cur.execute(Schema.ADD_PROGRESS_PCT)
+            await cur.execute(Schema.ADD_PROGRESS_LABEL)
+            await cur.execute(Schema.ADD_PROGRESS_DETAIL)
+            await cur.execute(Schema.IDX_ANALYSIS_JOBS_CREATED_AT)
+            await cur.execute(Schema.IDX_ANALYSIS_JOBS_STATUS)
+            await cur.execute(Schema.ENSURE_MAILS)
+            await cur.execute(Schema.IDX_MAILS_JOB_ID)
+            await cur.execute(Schema.IDX_MAILS_JOB_MAILED_AT)
+            await cur.execute(Schema.IDX_MAILS_THREAD_KEY)
+            await cur.execute(Schema.IDX_MAILS_MESSAGE_ID)
+            await cur.execute(Schema.ENSURE_WEEKLY_REPORTS)
+            await cur.execute(Schema.IDX_WEEKLY_REPORTS_CREATED_AT)
+        await conn.commit()
+    logger.info("Ensured mail_summary tables")
+
+
+async def create_job(
+    *,
+    mailbox_scope: str,
+    date_from: date,
+    date_to: date,
+    model: str,
+    inbox_zip_name: str | None,
+    sent_zip_name: str | None,
+) -> dict[str, Any]:
+    if not pool_enabled():
+        raise RuntimeError("database_unavailable")
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(
+                Jobs.INSERT,
+                {
+                    "mailbox_scope": mailbox_scope,
+                    "date_from": date_from,
+                    "date_to": date_to,
+                    "model": model,
+                    "inbox_zip_name": inbox_zip_name,
+                    "sent_zip_name": sent_zip_name,
+                },
+            )
+            row = await cur.fetchone()
+        await conn.commit()
+    return _row_to_job(row)
+
+
+async def get_job(job_id: int) -> dict[str, Any] | None:
+    if not pool_enabled():
+        return None
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(Jobs.GET_BY_ID, {"job_id": job_id})
+            row = await cur.fetchone()
+    return _row_to_job(row) if row else None
+
+
+async def list_jobs(limit: int = 20) -> list[dict[str, Any]]:
+    if not pool_enabled():
+        return []
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(Jobs.LIST_RECENT, {"limit": limit})
+            rows = await cur.fetchall()
+    return [_row_to_job(row) for row in rows]
+
+
+async def update_job(
+    job_id: int,
+    *,
+    status: str | None = None,
+    inbox_count: int | None = None,
+    sent_count: int | None = None,
+    filtered_inbox_count: int | None = None,
+    filtered_sent_count: int | None = None,
+    progress_pct: int | None = None,
+    progress_label: str | None = None,
+    progress_detail: str | None = None,
+    error_message: str | None = None,
+    started: bool = False,
+    completed: bool = False,
+) -> dict[str, Any] | None:
+    if not pool_enabled():
+        return None
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(
+                Jobs.UPDATE,
+                {
+                    "job_id": job_id,
+                    "status": status,
+                    "inbox_count": inbox_count,
+                    "sent_count": sent_count,
+                    "filtered_inbox_count": filtered_inbox_count,
+                    "filtered_sent_count": filtered_sent_count,
+                    "progress_pct": progress_pct,
+                    "progress_label": progress_label,
+                    "progress_detail": progress_detail,
+                    "error_message": error_message,
+                    "set_started": started,
+                    "set_completed": completed,
+                },
+            )
+            row = await cur.fetchone()
+        await conn.commit()
+    return _row_to_job(row) if row else None
+
+
+async def insert_mails(job_id: int, mails: list[ParsedMail]) -> int:
+    if not pool_enabled() or not mails:
+        return 0
+    pool = get_pool()
+    count = 0
+    async with pool.connection() as conn:
+        async with conn.cursor() as cur:
+            for mail in mails:
+                await cur.execute(
+                    Mails.INSERT,
+                    {
+                        "job_id": job_id,
+                        "mailbox": mail.mailbox,
+                        "message_id": mail.message_id,
+                        "thread_key": mail.thread_key,
+                        "subject": mail.subject,
+                        "from_addr": mail.from_addr,
+                        "to_addrs": mail.to_addrs,
+                        "cc_addrs": mail.cc_addrs,
+                        "mailed_at": mail.mailed_at,
+                        "attachment_names": Jsonb(mail.attachment_names),
+                        "body_text": mail.body_text,
+                    },
+                )
+                count += 1
+        await conn.commit()
+    return count
+
+
+async def list_mails_for_job(job_id: int) -> list[dict[str, Any]]:
+    if not pool_enabled():
+        return []
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(Mails.LIST_BY_JOB, {"job_id": job_id})
+            rows = await cur.fetchall()
+    result: list[dict[str, Any]] = []
+    for row in rows:
+        result.append(
+            {
+                "id": row["id"],
+                "job_id": row["job_id"],
+                "mailbox": row["mailbox"],
+                "message_id": row["message_id"],
+                "thread_key": row["thread_key"],
+                "subject": row["subject"],
+                "from_addr": row["from_addr"],
+                "to_addrs": row["to_addrs"],
+                "cc_addrs": row["cc_addrs"],
+                "mailed_at": row["mailed_at"].isoformat() if row["mailed_at"] else None,
+                "attachment_names": row["attachment_names"] or [],
+                "body_text": row["body_text"],
+                "body_purged": row["body_purged"],
+                "summary": row["summary"],
+            }
+        )
+    return result
+
+
+async def update_mail_summary(mail_id: int, summary: dict[str, Any]) -> None:
+    if not pool_enabled():
+        return
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                Mails.UPDATE_SUMMARY,
+                {
+                    "mail_id": mail_id,
+                    "summary": Jsonb(summary),
+                },
+            )
+        await conn.commit()
+
+
+async def purge_mail_bodies(job_id: int) -> None:
+    if not pool_enabled():
+        return
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(Mails.PURGE_BODIES, {"job_id": job_id})
+        await conn.commit()
+
+
+async def get_report(job_id: int) -> dict[str, Any] | None:
+    if not pool_enabled():
+        return None
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(Reports.GET_BY_JOB, {"job_id": job_id})
+            row = await cur.fetchone()
+    return _row_to_report(row) if row else None
+
+
+async def save_weekly_report(
+    *,
+    job_id: int,
+    model: str,
+    scope_note: str,
+    report: dict[str, Any],
+    report_markdown: str,
+) -> dict[str, Any]:
+    if not pool_enabled():
+        raise RuntimeError("database_unavailable")
+    pool = get_pool()
+    async with pool.connection() as conn:
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(
+                Reports.UPSERT,
+                {
+                    "job_id": job_id,
+                    "model": model,
+                    "scope_note": scope_note,
+                    "report": Jsonb(report),
+                    "report_markdown": report_markdown,
+                },
+            )
+            row = await cur.fetchone()
+        await conn.commit()
+    return _row_to_report(row)

backend/llm_client.py

@@ -0,0 +1,93 @@
+"""Local Ollama client for mail analysis."""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from typing import Any
+
+import httpx
+
+from config import LLM_BASE_URL, LLM_MODEL, LLM_TIMEOUT_SECONDS
+
+logger = logging.getLogger("mail_summary")
+
+_JSON_BLOCK_RE = re.compile(r"\{.*\}", re.DOTALL)
+
+
+class LlmError(RuntimeError):
+    pass
+
+
+def extract_json_object(text: str) -> dict[str, Any]:
+    text = (text or "").strip()
+    if not text:
+        raise LlmError("empty_llm_response")
+    try:
+        data = json.loads(text)
+        if isinstance(data, dict):
+            return data
+    except json.JSONDecodeError:
+        pass
+    match = _JSON_BLOCK_RE.search(text)
+    if not match:
+        raise LlmError("llm_response_not_json")
+    data = json.loads(match.group(0))
+    if not isinstance(data, dict):
+        raise LlmError("llm_response_not_object")
+    return data
+
+
+async def chat_json(
+    *,
+    system: str,
+    user: str,
+    model: str | None = None,
+    temperature: float = 0.2,
+) -> dict[str, Any]:
+    selected = (model or LLM_MODEL).strip() or LLM_MODEL
+    payload = {
+        "model": selected,
+        "stream": False,
+        "format": "json",
+        "options": {"temperature": temperature},
+        "messages": [
+            {"role": "system", "content": system},
+            {"role": "user", "content": user},
+        ],
+    }
+    url = f"{LLM_BASE_URL}/api/chat"
+    try:
+        async with httpx.AsyncClient(timeout=LLM_TIMEOUT_SECONDS) as client:
+            response = await client.post(url, json=payload)
+            response.raise_for_status()
+            body = response.json()
+    except httpx.HTTPError as exc:
+        logger.exception("Ollama request failed")
+        raise LlmError(f"ollama_request_failed: {exc}") from exc
+
+    content = (
+        (body.get("message") or {}).get("content")
+        or body.get("response")
+        or ""
+    )
+    return extract_json_object(content)
+
+
+async def ping_llm(model: str | None = None) -> dict[str, Any]:
+    selected = (model or LLM_MODEL).strip() or LLM_MODEL
+    try:
+        async with httpx.AsyncClient(timeout=3.0) as client:
+            tags = await client.get(f"{LLM_BASE_URL}/api/tags")
+            tags.raise_for_status()
+            data = tags.json()
+    except Exception as exc:  # noqa: BLE001
+        return {"ok": False, "error": str(exc), "model": selected}
+    names = [m.get("name", "") for m in data.get("models", [])]
+    return {
+        "ok": True,
+        "model": selected,
+        "model_present": any(selected in name or name in selected for name in names),
+        "models": names[:20],
+    }

backend/mail_parser.py

@@ -0,0 +1,342 @@
+"""Zip(.eml) extraction, parsing, cleaning, and date filtering."""
+
+from __future__ import annotations
+
+import email
+import email.policy
+import re
+import zipfile
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from datetime import date, datetime, timezone
+from email.header import decode_header, make_header
+from email.message import EmailMessage
+from email.utils import parsedate_to_datetime
+from html.parser import HTMLParser
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+KST = ZoneInfo("Asia/Seoul")
+
+_SUBJECT_PREFIX_RE = re.compile(
+    r"^\s*((re|fw|fwd|답장|전달|회신)\s*[:：]|\[?\s*(re|fw|fwd)\s*\]?\s*:)\s*",
+    re.IGNORECASE,
+)
+_QUOTE_MARKERS = (
+    "-----original message-----",
+    "----- 원본 메시지 -----",
+    "-----원본 메시지-----",
+    "________________________________",
+    "begin forwarded message",
+    "보낸 사람:",
+    "보내는 사람:",
+    "보내는사람:",
+    "보내는사람 :",
+    "from:",
+)
+_SIG_MARKERS = (
+    "\n-- \n",
+    "\n--\n",
+    "\n감사합니다.\n",
+    "\nthanks,\n",
+    "\nbest regards,\n",
+    "\ncustomer centric agility\n",
+    "\ntel :",
+    "\ntel:",
+)
+
+
+@dataclass
+class ParsedMail:
+    mailbox: str
+    message_id: str
+    thread_key: str
+    subject: str
+    from_addr: str
+    to_addrs: str
+    cc_addrs: str
+    mailed_at: datetime | None
+    attachment_names: list[str] = field(default_factory=list)
+    body_text: str = ""
+    source_path: str = ""
+
+
+class _HTMLTextExtractor(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__()
+        self._parts: list[str] = []
+        self._skip = False
+
+    def handle_starttag(self, tag: str, attrs) -> None:  # noqa: ANN001
+        if tag in {"script", "style"}:
+            self._skip = True
+        elif tag in {"br", "p", "div", "tr", "li"}:
+            self._parts.append("\n")
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag in {"script", "style"}:
+            self._skip = False
+        elif tag in {"p", "div", "tr", "li"}:
+            self._parts.append("\n")
+
+    def handle_data(self, data: str) -> None:
+        if not self._skip:
+            self._parts.append(data)
+
+    def text(self) -> str:
+        return "".join(self._parts)
+
+
+def decode_mime_header(value: str | None) -> str:
+    if not value:
+        return ""
+    try:
+        return str(make_header(decode_header(value))).strip()
+    except Exception:  # noqa: BLE001
+        return value.strip()
+
+
+def normalize_thread_key(subject: str) -> str:
+    text = decode_mime_header(subject)
+    prev = None
+    while prev != text:
+        prev = text
+        text = _SUBJECT_PREFIX_RE.sub("", text).strip()
+    text = re.sub(r"\s+", " ", text).strip().lower()
+    return text or "(no-subject)"
+
+
+def html_to_text(html: str) -> str:
+    parser = _HTMLTextExtractor()
+    try:
+        parser.feed(html)
+        parser.close()
+    except Exception:  # noqa: BLE001
+        return re.sub(r"<[^>]+>", " ", html)
+    return parser.text()
+
+
+def clean_body(text: str) -> str:
+    if not text:
+        return ""
+    normalized = text.replace("\r\n", "\n").replace("\r", "\n")
+    lower = normalized.lower()
+    cut = len(normalized)
+    for marker in _QUOTE_MARKERS:
+        idx = lower.find(marker)
+        if idx != -1:
+            cut = min(cut, idx)
+    body = normalized[:cut]
+    for marker in _SIG_MARKERS:
+        idx = body.lower().find(marker.lower())
+        if idx != -1 and idx > 40:
+            body = body[:idx]
+            break
+
+    lines: list[str] = []
+    for line in body.split("\n"):
+        stripped = line.strip()
+        if stripped.startswith(">"):
+            continue
+        if re.match(
+            r"^(from|sent|to|cc|subject|보내는\s*사람|보내는사람)\s*:",
+            stripped,
+            re.IGNORECASE,
+        ):
+            # Likely start of an inline quoted header block.
+            break
+        if re.match(r"^customer centric agility$", stripped, re.IGNORECASE):
+            break
+        if re.match(r"^tel\s*:", stripped, re.IGNORECASE):
+            break
+        lines.append(line.rstrip())
+
+    cleaned = "\n".join(lines)
+    cleaned = re.sub(r"\n{3,}", "\n\n", cleaned).strip()
+    return cleaned
+
+
+def _addresses(msg: EmailMessage, header: str) -> str:
+    raw = msg.get_all(header, [])
+    if not raw:
+        return ""
+    values = [decode_mime_header(str(item)) for item in raw]
+    return ", ".join(v for v in values if v)
+
+
+def _parse_date(msg: EmailMessage) -> datetime | None:
+    raw = msg.get("Date")
+    if not raw:
+        return None
+    try:
+        dt = parsedate_to_datetime(raw)
+    except Exception:  # noqa: BLE001
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(KST)
+
+
+def _extract_body(msg: EmailMessage) -> str:
+    text_part = ""
+    html_part = ""
+    if msg.is_multipart():
+        for part in msg.walk():
+            ctype = part.get_content_type()
+            disp = str(part.get("Content-Disposition") or "")
+            if "attachment" in disp.lower():
+                continue
+            try:
+                payload = part.get_content()
+            except Exception:  # noqa: BLE001
+                continue
+            if not isinstance(payload, str):
+                continue
+            if ctype == "text/plain" and not text_part:
+                text_part = payload
+            elif ctype == "text/html" and not html_part:
+                html_part = payload
+    else:
+        try:
+            payload = msg.get_content()
+        except Exception:  # noqa: BLE001
+            payload = ""
+        if isinstance(payload, str):
+            if msg.get_content_type() == "text/html":
+                html_part = payload
+            else:
+                text_part = payload
+
+    raw = text_part if text_part.strip() else html_to_text(html_part)
+    return clean_body(raw)
+
+
+def _attachment_names(msg: EmailMessage) -> list[str]:
+    names: list[str] = []
+    for part in msg.walk():
+        filename = part.get_filename()
+        if filename:
+            names.append(decode_mime_header(filename))
+    return names
+
+
+def parse_eml_bytes(data: bytes, mailbox: str, source_name: str) -> ParsedMail | None:
+    try:
+        msg = email.message_from_bytes(data, policy=email.policy.default)
+    except Exception:  # noqa: BLE001
+        return None
+
+    subject = decode_mime_header(msg.get("Subject"))
+    message_id = decode_mime_header(msg.get("Message-ID")) or Path(source_name).name
+    return ParsedMail(
+        mailbox=mailbox,
+        message_id=message_id,
+        thread_key=normalize_thread_key(subject),
+        subject=subject,
+        from_addr=_addresses(msg, "From"),
+        to_addrs=_addresses(msg, "To"),
+        cc_addrs=_addresses(msg, "Cc"),
+        mailed_at=_parse_date(msg),
+        attachment_names=_attachment_names(msg),
+        body_text=_extract_body(msg),
+        source_path=source_name,
+    )
+
+
+def parse_eml_file(path: Path, mailbox: str) -> ParsedMail | None:
+    try:
+        return parse_eml_bytes(path.read_bytes(), mailbox, str(path))
+    except Exception:  # noqa: BLE001
+        return None
+
+
+def find_zip(job_dir: Path, prefix: str) -> Path | None:
+    matches = sorted(job_dir.glob(f"{prefix}_*.zip"))
+    return matches[0] if matches else None
+
+
+def iter_eml_members(zip_path: Path) -> list[zipfile.ZipInfo]:
+    members: list[zipfile.ZipInfo] = []
+    with zipfile.ZipFile(zip_path, "r") as zf:
+        for info in zf.infolist():
+            if info.is_dir():
+                continue
+            name = info.filename.replace("\\", "/")
+            if name.startswith("/") or ".." in Path(name).parts:
+                continue
+            if not name.lower().endswith(".eml"):
+                continue
+            members.append(info)
+    return members
+
+
+def load_mailbox_from_zip(
+    zip_path: Path,
+    mailbox: str,
+    *,
+    date_from: date | None = None,
+    date_to: date | None = None,
+    on_progress: Callable[[int, int, int], None] | None = None,
+) -> tuple[list[ParsedMail], int]:
+    """Stream-parse .eml entries inside zip. Optionally keep only mails in date range.
+
+    Returns (mails, total_eml_count). Avoids extracting multi-GB archives to disk.
+    on_progress(scanned, total, matched) is called periodically.
+    """
+    members = iter_eml_members(zip_path)
+    total = len(members)
+    mails: list[ParsedMail] = []
+    scanned = 0
+
+    if on_progress:
+        on_progress(0, total, 0)
+
+    with zipfile.ZipFile(zip_path, "r") as zf:
+        for info in members:
+            scanned += 1
+            name = info.filename.replace("\\", "/")
+            try:
+                with zf.open(info) as src:
+                    data = src.read()
+            except Exception:  # noqa: BLE001
+                data = b""
+            if data:
+                parsed = parse_eml_bytes(data, mailbox, name)
+                if parsed:
+                    keep = True
+                    if date_from and date_to:
+                        if parsed.mailed_at is None:
+                            keep = False
+                        else:
+                            day = parsed.mailed_at.astimezone(KST).date()
+                            keep = date_from <= day <= date_to
+                    if keep:
+                        mails.append(parsed)
+
+            if on_progress and (scanned == total or scanned % 25 == 0):
+                on_progress(scanned, total, len(mails))
+
+    mails.sort(key=lambda m: m.mailed_at or datetime.min.replace(tzinfo=KST))
+    if on_progress:
+        on_progress(total, total, len(mails))
+    return mails, total
+
+def filter_by_date(mails: list[ParsedMail], date_from: date, date_to: date) -> list[ParsedMail]:
+    filtered: list[ParsedMail] = []
+    for mail in mails:
+        if mail.mailed_at is None:
+            continue
+        local_day = mail.mailed_at.astimezone(KST).date()
+        if date_from <= local_day <= date_to:
+            filtered.append(mail)
+    filtered.sort(key=lambda m: m.mailed_at or datetime.min.replace(tzinfo=KST))
+    return filtered
+
+
+def group_by_thread(mails: list[ParsedMail]) -> dict[str, list[ParsedMail]]:
+    groups: dict[str, list[ParsedMail]] = {}
+    for mail in mails:
+        groups.setdefault(mail.thread_key, []).append(mail)
+    for key in groups:
+        groups[key].sort(key=lambda m: m.mailed_at or datetime.min.replace(tzinfo=KST))
+    return groups

backend/pipeline.py

@@ -0,0 +1,768 @@
+"""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("&nbsp;", " ")
+    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,
+        )

backend/queries.py

@@ -0,0 +1,232 @@
+"""Centralized SQL for mail_summary."""
+
+
+class Schema:
+    ENSURE_ANALYSIS_JOBS = """
+    CREATE TABLE IF NOT EXISTS analysis_jobs (
+        id                   BIGSERIAL PRIMARY KEY,
+        status               TEXT NOT NULL DEFAULT 'pending',
+        mailbox_scope        TEXT NOT NULL DEFAULT 'both',
+        date_from            DATE NOT NULL,
+        date_to              DATE NOT NULL,
+        model                TEXT NOT NULL DEFAULT '',
+        inbox_zip_name       TEXT,
+        sent_zip_name        TEXT,
+        inbox_count          INTEGER NOT NULL DEFAULT 0,
+        sent_count           INTEGER NOT NULL DEFAULT 0,
+        filtered_inbox_count INTEGER NOT NULL DEFAULT 0,
+        filtered_sent_count  INTEGER NOT NULL DEFAULT 0,
+        progress_pct         INTEGER NOT NULL DEFAULT 0,
+        progress_label       TEXT NOT NULL DEFAULT '',
+        progress_detail      TEXT NOT NULL DEFAULT '',
+        error_message        TEXT,
+        created_at           TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        started_at           TIMESTAMPTZ,
+        completed_at         TIMESTAMPTZ,
+        CONSTRAINT analysis_jobs_status_chk CHECK (
+            status IN ('pending', 'extracting', 'filtering', 'analyzing', 'completed', 'failed')
+        ),
+        CONSTRAINT analysis_jobs_scope_chk CHECK (
+            mailbox_scope IN ('inbox', 'sent', 'both')
+        ),
+        CONSTRAINT analysis_jobs_date_chk CHECK (date_from <= date_to)
+    )
+    """
+    ADD_PROGRESS_PCT = """
+    ALTER TABLE analysis_jobs
+    ADD COLUMN IF NOT EXISTS progress_pct INTEGER NOT NULL DEFAULT 0
+    """
+    ADD_PROGRESS_LABEL = """
+    ALTER TABLE analysis_jobs
+    ADD COLUMN IF NOT EXISTS progress_label TEXT NOT NULL DEFAULT ''
+    """
+    ADD_PROGRESS_DETAIL = """
+    ALTER TABLE analysis_jobs
+    ADD COLUMN IF NOT EXISTS progress_detail TEXT NOT NULL DEFAULT ''
+    """
+    IDX_ANALYSIS_JOBS_CREATED_AT = """
+    CREATE INDEX IF NOT EXISTS idx_analysis_jobs_created_at
+        ON analysis_jobs (created_at DESC)
+    """
+    IDX_ANALYSIS_JOBS_STATUS = """
+    CREATE INDEX IF NOT EXISTS idx_analysis_jobs_status
+        ON analysis_jobs (status)
+    """
+
+    ENSURE_MAILS = """
+    CREATE TABLE IF NOT EXISTS mails (
+        id                BIGSERIAL PRIMARY KEY,
+        job_id            BIGINT NOT NULL REFERENCES analysis_jobs(id) ON DELETE CASCADE,
+        mailbox           TEXT NOT NULL,
+        message_id        TEXT,
+        thread_key        TEXT,
+        subject           TEXT NOT NULL DEFAULT '',
+        from_addr         TEXT NOT NULL DEFAULT '',
+        to_addrs          TEXT NOT NULL DEFAULT '',
+        cc_addrs          TEXT NOT NULL DEFAULT '',
+        mailed_at         TIMESTAMPTZ,
+        attachment_names  JSONB NOT NULL DEFAULT '[]'::jsonb,
+        body_text         TEXT,
+        body_purged       BOOLEAN NOT NULL DEFAULT FALSE,
+        summary           JSONB,
+        created_at        TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        CONSTRAINT mails_mailbox_chk CHECK (mailbox IN ('inbox', 'sent'))
+    )
+    """
+
+    IDX_MAILS_JOB_ID = "CREATE INDEX IF NOT EXISTS idx_mails_job_id ON mails (job_id)"
+    IDX_MAILS_JOB_MAILED_AT = (
+        "CREATE INDEX IF NOT EXISTS idx_mails_job_mailed_at ON mails (job_id, mailed_at)"
+    )
+    IDX_MAILS_THREAD_KEY = (
+        "CREATE INDEX IF NOT EXISTS idx_mails_thread_key ON mails (job_id, thread_key)"
+    )
+    IDX_MAILS_MESSAGE_ID = (
+        "CREATE INDEX IF NOT EXISTS idx_mails_message_id ON mails (job_id, message_id)"
+    )
+
+    ENSURE_WEEKLY_REPORTS = """
+    CREATE TABLE IF NOT EXISTS weekly_reports (
+        id              BIGSERIAL PRIMARY KEY,
+        job_id          BIGINT NOT NULL UNIQUE REFERENCES analysis_jobs(id) ON DELETE CASCADE,
+        model           TEXT NOT NULL DEFAULT '',
+        scope_note      TEXT NOT NULL DEFAULT '',
+        report          JSONB NOT NULL DEFAULT '{}'::jsonb,
+        report_markdown TEXT,
+        created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    )
+    """
+
+    IDX_WEEKLY_REPORTS_CREATED_AT = """
+    CREATE INDEX IF NOT EXISTS idx_weekly_reports_created_at
+        ON weekly_reports (created_at DESC)
+    """
+
+class Jobs:
+    INSERT = """
+    INSERT INTO analysis_jobs (
+        status, mailbox_scope, date_from, date_to, model,
+        inbox_zip_name, sent_zip_name
+    )
+    VALUES (
+        'pending', %(mailbox_scope)s, %(date_from)s, %(date_to)s, %(model)s,
+        %(inbox_zip_name)s, %(sent_zip_name)s
+    )
+    RETURNING id, status, mailbox_scope, date_from, date_to, model,
+              inbox_zip_name, sent_zip_name,
+              inbox_count, sent_count, filtered_inbox_count, filtered_sent_count,
+              progress_pct, progress_label, progress_detail,
+              error_message, created_at, started_at, completed_at
+    """
+
+    GET_BY_ID = """
+    SELECT id, status, mailbox_scope, date_from, date_to, model,
+           inbox_zip_name, sent_zip_name,
+           inbox_count, sent_count, filtered_inbox_count, filtered_sent_count,
+           progress_pct, progress_label, progress_detail,
+           error_message, created_at, started_at, completed_at
+    FROM analysis_jobs
+    WHERE id = %(job_id)s
+    """
+
+    LIST_RECENT = """
+    SELECT id, status, mailbox_scope, date_from, date_to, model,
+           inbox_zip_name, sent_zip_name,
+           inbox_count, sent_count, filtered_inbox_count, filtered_sent_count,
+           progress_pct, progress_label, progress_detail,
+           error_message, created_at, started_at, completed_at
+    FROM analysis_jobs
+    ORDER BY created_at DESC
+    LIMIT %(limit)s
+    """
+
+    UPDATE = """
+    UPDATE analysis_jobs
+    SET
+        status = COALESCE(%(status)s, status),
+        inbox_count = COALESCE(%(inbox_count)s, inbox_count),
+        sent_count = COALESCE(%(sent_count)s, sent_count),
+        filtered_inbox_count = COALESCE(%(filtered_inbox_count)s, filtered_inbox_count),
+        filtered_sent_count = COALESCE(%(filtered_sent_count)s, filtered_sent_count),
+        progress_pct = COALESCE(%(progress_pct)s, progress_pct),
+        progress_label = COALESCE(%(progress_label)s, progress_label),
+        progress_detail = COALESCE(%(progress_detail)s, progress_detail),
+        error_message = COALESCE(%(error_message)s, error_message),
+        started_at = CASE
+            WHEN %(set_started)s THEN COALESCE(started_at, NOW())
+            ELSE started_at
+        END,
+        completed_at = CASE
+            WHEN %(set_completed)s THEN NOW()
+            ELSE completed_at
+        END
+    WHERE id = %(job_id)s
+    RETURNING id, status, mailbox_scope, date_from, date_to, model,
+              inbox_zip_name, sent_zip_name,
+              inbox_count, sent_count, filtered_inbox_count, filtered_sent_count,
+              progress_pct, progress_label, progress_detail,
+              error_message, created_at, started_at, completed_at
+    """
+
+
+class Mails:
+    INSERT = """
+    INSERT INTO mails (
+        job_id, mailbox, message_id, thread_key, subject,
+        from_addr, to_addrs, cc_addrs, mailed_at,
+        attachment_names, body_text
+    )
+    VALUES (
+        %(job_id)s, %(mailbox)s, %(message_id)s, %(thread_key)s, %(subject)s,
+        %(from_addr)s, %(to_addrs)s, %(cc_addrs)s, %(mailed_at)s,
+        %(attachment_names)s, %(body_text)s
+    )
+    RETURNING id
+    """
+
+    LIST_BY_JOB = """
+    SELECT id, job_id, mailbox, message_id, thread_key, subject,
+           from_addr, to_addrs, cc_addrs, mailed_at,
+           attachment_names, body_text, body_purged, summary, created_at
+    FROM mails
+    WHERE job_id = %(job_id)s
+    ORDER BY mailed_at NULLS LAST, id
+    """
+
+    UPDATE_SUMMARY = """
+    UPDATE mails
+    SET summary = %(summary)s
+    WHERE id = %(mail_id)s
+    """
+
+    PURGE_BODIES = """
+    UPDATE mails
+    SET body_text = NULL, body_purged = TRUE
+    WHERE job_id = %(job_id)s
+    """
+
+
+class Reports:
+    GET_BY_JOB = """
+    SELECT id, job_id, model, scope_note, report, report_markdown,
+           created_at, updated_at
+    FROM weekly_reports
+    WHERE job_id = %(job_id)s
+    """
+
+    UPSERT = """
+    INSERT INTO weekly_reports (job_id, model, scope_note, report, report_markdown)
+    VALUES (
+        %(job_id)s, %(model)s, %(scope_note)s,
+        %(report)s, %(report_markdown)s
+    )
+    ON CONFLICT (job_id) DO UPDATE SET
+        model = EXCLUDED.model,
+        scope_note = EXCLUDED.scope_note,
+        report = EXCLUDED.report,
+        report_markdown = EXCLUDED.report_markdown,
+        updated_at = NOW()
+    RETURNING id, job_id, model, scope_note, report, report_markdown,
+              created_at, updated_at
+    """

backend/requirements.txt

@@ -0,0 +1,7 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.32.0
+python-multipart>=0.0.12
+httpx>=0.27.0
+python-dotenv>=1.0.0
+psycopg[binary,pool]>=3.2.0
+eval_type_backport>=0.4.0