383 lines
13 KiB
Python
383 lines
13 KiB
Python
"""数据导出（Excel / xlsx）。
|
||
|
||
提供四类导出，均可带上页面当前的筛选条件：
|
||
- /export/custom 定制项目明细
|
||
- /export/custom-updates 定制项目周期记录
|
||
- /export/projects 常规项目管理
|
||
- /export/updates 常规项目周期执行记录
|
||
- /export/staff 人力资源库（含项目负载）
|
||
- /export/focus 每周重点工作
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from fastapi.responses import Response
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..constants import CUSTOM_STAGE_MAP, STATUS_MAP
|
||
from ..db import get_db
|
||
from ..export import build_xlsx, xlsx_response
|
||
from ..models import (
|
||
CustomProject,
|
||
CustomUpdate,
|
||
Period,
|
||
Project,
|
||
Staff,
|
||
Task,
|
||
Update,
|
||
WeeklyFocus,
|
||
)
|
||
from .board import worst_status
|
||
from .custom import _sort_by_entry, _sort_by_update
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def _stage_name(code: str | None) -> str:
|
||
return CUSTOM_STAGE_MAP.get(code or "", (code or "", "", 0))[0]
|
||
|
||
|
||
def _resp(content: bytes, filename: str) -> Response:
|
||
body, headers = xlsx_response(content, filename)
|
||
return Response(content=body, headers=headers)
|
||
|
||
|
||
# ---------------------------------------------------------------- 定制项目
|
||
@router.get("/export/custom")
|
||
def export_custom(
|
||
region: str | None = None,
|
||
origin: str | None = None,
|
||
status: str | None = None,
|
||
risk: str | None = None,
|
||
accept_plan: str | None = None,
|
||
keyword: str | None = None,
|
||
archived: str = Query("false", pattern="^(true|false|all)$"),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
q = db.query(CustomProject)
|
||
if archived == "false":
|
||
q = q.filter(CustomProject.archived.is_(False))
|
||
elif archived == "true":
|
||
q = q.filter(CustomProject.archived.is_(True))
|
||
if region:
|
||
q = q.filter(CustomProject.region == region)
|
||
if origin:
|
||
q = q.filter(CustomProject.origin == origin)
|
||
if status:
|
||
q = q.filter(CustomProject.project_status == status)
|
||
if risk:
|
||
q = q.filter(CustomProject.risk_level == risk)
|
||
if accept_plan:
|
||
q = q.filter(CustomProject.accept_plan == accept_plan)
|
||
if keyword:
|
||
q = q.filter(CustomProject.name.contains(keyword))
|
||
projects = _sort_by_update(q.all())
|
||
|
||
# 列与「定制项目明细」表格一致
|
||
headers = [
|
||
"项目归属", "项目名称", "项目ID", "下单时间", "办事处", "行业",
|
||
"市场下单人天", "订单金额(元)", "市场责任人", "项目状态",
|
||
"验收时间", "推动验收计划", "风险值", "备注", "更新时间",
|
||
]
|
||
rows = []
|
||
for c in projects:
|
||
rows.append(
|
||
[
|
||
c.origin or "",
|
||
c.name,
|
||
c.project_code or "",
|
||
c.sign_date or "",
|
||
c.office or "",
|
||
c.industry or "",
|
||
c.man_days if c.man_days is not None else "",
|
||
c.amount if c.amount is not None else "",
|
||
c.market_owner or "",
|
||
c.project_status or "",
|
||
c.accept_date or "",
|
||
c.accept_plan or "",
|
||
c.risk_level or "",
|
||
(c.remark or "").replace("\n", " "),
|
||
c.update_date or "",
|
||
]
|
||
)
|
||
|
||
dicts = [c.to_dict() for c in projects]
|
||
|
||
def total(key: str) -> float:
|
||
return round(sum(d.get(key) or 0 for d in dicts), 2)
|
||
|
||
rows.append([""] * 15)
|
||
rows.append(
|
||
["合计", f"{len(projects)} 个项目", "", "", "", "",
|
||
total("man_days"), total("amount"), "", "", "", "", "", "", ""]
|
||
)
|
||
|
||
content = build_xlsx(
|
||
"定制项目明细",
|
||
headers,
|
||
rows,
|
||
widths=[10, 46, 18, 12, 10, 8, 13, 14, 14, 11, 11, 14, 10, 40, 18],
|
||
title="定制项目明细",
|
||
money_cols=(8,), # 订单金额(元)
|
||
)
|
||
return _resp(content, "定制项目明细")
|
||
|
||
|
||
@router.get("/export/custom-updates")
|
||
def export_custom_updates(
|
||
period_id: int | None = None,
|
||
year: int | None = None,
|
||
month: int | None = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
from .board import resolve_period_ids
|
||
|
||
period_ids = resolve_period_ids(db, period_id, year, month)
|
||
q = db.query(CustomUpdate)
|
||
if period_ids:
|
||
q = q.filter(CustomUpdate.period_id.in_(period_ids))
|
||
ups = q.order_by(CustomUpdate.period_id, CustomUpdate.custom_project_id).all()
|
||
|
||
headers = [
|
||
"周期", "项目名称", "区域", "阶段", "执行情况", "进度(%)",
|
||
"本期新增下单(元)", "本期新增计收(元)", "风险", "下一步",
|
||
]
|
||
rows = []
|
||
for u in ups:
|
||
cp = u.project
|
||
rows.append(
|
||
[
|
||
u.period.to_dict()["full_label"] if u.period else "",
|
||
cp.name if cp else "",
|
||
(cp.region or "") if cp else "",
|
||
_stage_name(u.stage),
|
||
(u.content or "").replace("\n", " ")[:500],
|
||
u.progress,
|
||
u.amount_delta,
|
||
u.revenue_delta,
|
||
(u.risk or "").replace("\n", " ")[:200],
|
||
(u.next_step or "").replace("\n", " ")[:200],
|
||
]
|
||
)
|
||
|
||
content = build_xlsx(
|
||
"定制项目记录",
|
||
headers,
|
||
rows,
|
||
widths=[18, 42, 10, 14, 70, 10, 16, 16, 30, 30],
|
||
title="定制项目周期记录",
|
||
money_cols=(7, 8),
|
||
)
|
||
return _resp(content, "定制项目周期记录")
|
||
|
||
|
||
# ---------------------------------------------------------------- 常规项目
|
||
@router.get("/export/projects")
|
||
def export_projects(
|
||
category_id: int | None = None,
|
||
keyword: str | None = None,
|
||
archived: str = Query("all", pattern="^(true|false|all)$"),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
q = db.query(Project)
|
||
if archived == "false":
|
||
q = q.filter(Project.archived.is_(False))
|
||
elif archived == "true":
|
||
q = q.filter(Project.archived.is_(True))
|
||
if category_id:
|
||
q = q.filter(Project.category_id == category_id)
|
||
if keyword:
|
||
q = q.filter(Project.name.contains(keyword))
|
||
projects = q.order_by(Project.is_key.desc(), Project.name).all()
|
||
|
||
periods = {p.id: p for p in db.query(Period).all()}
|
||
headers = [
|
||
"项目名称", "分类", "负责人", "状态", "优先级", "重点",
|
||
"进度(%)", "标签", "子任务数", "执行记录数", "最近更新周期", "归档",
|
||
]
|
||
status_names = {
|
||
"not_started": "未开始", "in_progress": "进行中", "done": "已完成",
|
||
"at_risk": "风险", "blocked": "阻塞", "paused": "暂停/终止",
|
||
}
|
||
rows = []
|
||
for p in projects:
|
||
ups = sorted(p.updates, key=lambda u: u.period_id or 0, reverse=True)
|
||
last = periods.get(ups[0].period_id) if ups else None
|
||
rows.append(
|
||
[
|
||
p.name,
|
||
p.category.name if p.category else "未分类",
|
||
p.owner or "",
|
||
status_names.get(p.status, p.status),
|
||
(p.priority or "").upper(),
|
||
"是" if p.is_key else "",
|
||
round(p.progress or 0, 1),
|
||
"、".join(t.name for t in p.tags),
|
||
len(p.tasks),
|
||
len(p.updates),
|
||
last.to_dict()["full_label"] if last else "",
|
||
"是" if p.archived else "",
|
||
]
|
||
)
|
||
|
||
content = build_xlsx(
|
||
"项目管理",
|
||
headers,
|
||
rows,
|
||
widths=[46, 12, 12, 10, 9, 7, 10, 24, 11, 12, 18, 7],
|
||
title="项目管理清单",
|
||
)
|
||
return _resp(content, "项目管理清单")
|
||
|
||
|
||
@router.get("/export/updates")
|
||
def export_updates(
|
||
period_id: int | None = None,
|
||
year: int | None = None,
|
||
month: int | None = None,
|
||
project_id: int | None = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
from .board import resolve_period_ids
|
||
|
||
period_ids = resolve_period_ids(db, period_id, year, month)
|
||
q = db.query(Update)
|
||
if period_ids:
|
||
q = q.filter(Update.period_id.in_(period_ids))
|
||
if project_id:
|
||
q = q.filter(Update.project_id == project_id)
|
||
ups = q.order_by(Update.period_id, Update.project_id, Update.id).all()
|
||
|
||
status_names = {
|
||
"not_started": "未开始", "in_progress": "进行中", "done": "已完成",
|
||
"at_risk": "风险", "blocked": "阻塞", "paused": "暂停/终止",
|
||
}
|
||
headers = [
|
||
"周期", "项目", "子任务", "执行情况", "进度(%)", "状态", "风险", "下一步", "来源",
|
||
]
|
||
rows = []
|
||
for u in ups:
|
||
rows.append(
|
||
[
|
||
u.period.to_dict()["full_label"] if u.period else "",
|
||
u.project.name if u.project else "",
|
||
u.task.name if u.task else "",
|
||
(u.content or "").replace("\n", " ")[:500],
|
||
u.progress,
|
||
status_names.get(u.status, u.status),
|
||
(u.risk or "").replace("\n", " ")[:200],
|
||
(u.next_step or "").replace("\n", " ")[:200],
|
||
{"import": "导入", "ai": "AI整理", "manual": "手工"}.get(u.source, u.source),
|
||
]
|
||
)
|
||
|
||
content = build_xlsx(
|
||
"周期执行记录",
|
||
headers,
|
||
rows,
|
||
widths=[18, 30, 22, 70, 10, 10, 30, 30, 10],
|
||
title="周期执行记录",
|
||
)
|
||
return _resp(content, "周期执行记录")
|
||
|
||
|
||
# ---------------------------------------------------------------- 人力资源库
|
||
@router.get("/export/staff")
|
||
def export_staff(
|
||
archived: str = Query("false", pattern="^(true|false|all)$"),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
from .staff import _workload
|
||
|
||
q = db.query(Staff)
|
||
if archived == "false":
|
||
q = q.filter(Staff.archived.is_(False))
|
||
elif archived == "true":
|
||
q = q.filter(Staff.archived.is_(True))
|
||
members = q.order_by(Staff.sort_order, Staff.id).all()
|
||
load = _workload(db)
|
||
|
||
headers = [
|
||
"序号", "姓名", "归属小组", "人员归属", "在管项目", "重点项目",
|
||
"定制项目", "定制合同额(元)", "定制待计收(元)", "最近更新", "说明", "归档",
|
||
]
|
||
rows = []
|
||
for i, m in enumerate(members, 1):
|
||
w = load.get(m.name, {})
|
||
rows.append(
|
||
[
|
||
i,
|
||
m.name,
|
||
m.group or "",
|
||
m.affiliation or "",
|
||
w.get("project_count", 0),
|
||
w.get("key_project_count", 0),
|
||
w.get("custom_count", 0),
|
||
w.get("custom_amount", 0),
|
||
w.get("custom_pending", 0),
|
||
w.get("last_update") or "",
|
||
(m.note or "").replace("\n", " "),
|
||
"是" if m.archived else "",
|
||
]
|
||
)
|
||
|
||
content = build_xlsx(
|
||
"人力资源库",
|
||
headers,
|
||
rows,
|
||
widths=[6, 12, 14, 10, 10, 10, 10, 16, 16, 16, 52, 7],
|
||
title="软件开发部人力资源库",
|
||
money_cols=(8, 9),
|
||
)
|
||
return _resp(content, "人力资源库")
|
||
|
||
|
||
# ---------------------------------------------------------------- 每周重点工作
|
||
@router.get("/export/focus")
|
||
def export_focus(
|
||
archived: str = Query("false", pattern="^(true|false|all)$"),
|
||
status: str | None = None,
|
||
keyword: str | None = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
q = db.query(WeeklyFocus)
|
||
if archived == "false":
|
||
q = q.filter(WeeklyFocus.archived.is_(False))
|
||
elif archived == "true":
|
||
q = q.filter(WeeklyFocus.archived.is_(True))
|
||
if status:
|
||
q = q.filter(WeeklyFocus.status == status)
|
||
if keyword:
|
||
q = q.filter(WeeklyFocus.content.contains(keyword) | WeeklyFocus.detail.contains(keyword))
|
||
periods = {p.id: p for p in db.query(Period).all()}
|
||
items = sorted(
|
||
q.all(),
|
||
key=lambda f: (-(periods[f.period_id].sort_key if f.period_id in periods else 0), -f.id),
|
||
)
|
||
|
||
headers = ["序号", "工作内容", "说明", "预计完成时间", "录入周", "执行状态", "是否延期", "结果反馈", "更新"]
|
||
rows = []
|
||
for i, f in enumerate(items, 1):
|
||
period = periods.get(f.period_id)
|
||
rows.append(
|
||
[
|
||
i,
|
||
f.content,
|
||
(f.detail or "").replace("\n", " "),
|
||
f.due_date or "",
|
||
period.to_dict()["full_label"] if period else "",
|
||
STATUS_MAP.get(f.status, (f.status,))[0],
|
||
"是" if f.delayed else "",
|
||
(f.result or "").replace("\n", " "),
|
||
f.updated_at.strftime("%Y-%m-%d %H:%M") if f.updated_at else "",
|
||
]
|
||
)
|
||
content = build_xlsx(
|
||
"每周重点工作",
|
||
headers,
|
||
rows,
|
||
widths=[6, 46, 54, 14, 14, 12, 10, 54, 17],
|
||
title="每周重点工作",
|
||
)
|
||
return _resp(content, "每周重点工作")
|