Files
company-ai-platform/app/modules/reports/chart.py
JiuContinent 9cf7c44393 ```
feat: 添加生命周期报告和AI规则管理功能

- 在Dockerfile中添加pillow依赖包用于图像处理
- 实现生命周期报告调度任务,支持日报和周报两种类型
- 新增TASK_RUN_LIFECYCLE任务常量和相关配置选项
- 扩展AI Agent服务以支持用户规则,并在分析时应用规则
- 添加AI用户规则创建、更新和查询接口
- 增加项目生命周期和财务需求分析技能
- 扩展现有模型以支持更完整的业务数据字段
- 实现飞书图片上传功能用于报告展示
```
2026-07-12 17:44:49 +08:00

183 lines
6.2 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from io import BytesIO
from typing import Any
from PIL import Image, ImageDraw, ImageFont
WIDTH = 1200
HEIGHT = 720
FINANCE_HEIGHT = 1200
BACKGROUND = (248, 250, 252)
FOREGROUND = (17, 24, 39)
MUTED = (75, 85, 99)
GRID = (209, 213, 219)
SERIES = ((37, 99, 235), (245, 158, 11), (220, 38, 38))
def render_lifecycle_chart(chart_data: dict[str, Any]) -> bytes:
finance = chart_data.get("finance") or {}
image = Image.new("RGB", (WIDTH, FINANCE_HEIGHT if finance else HEIGHT), BACKGROUND)
draw = ImageDraw.Draw(image)
title_font = _font(36)
section_font = _font(28)
label_font = _font(22)
value_font = _font(22)
draw.text((60, 35), "Lifecycle Analysis", fill=FOREGROUND, font=title_font)
draw.text((60, 82), str(chart_data.get("period") or ""), fill=MUTED, font=label_font)
projects = chart_data.get("projects") or {}
draw.text((60, 135), "Project Portfolio", fill=FOREGROUND, font=section_font)
_stacked_bar(
draw,
y=190,
values=[
("Unarchived", int(projects.get("unarchived") or 0), SERIES[0]),
("Archived", int(projects.get("archived") or 0), GRID),
],
total=max(int(projects.get("total") or 0), 1),
label_font=label_font,
value_font=value_font,
)
risks = chart_data.get("risks") or {}
draw.text((60, 295), "Current Risks", fill=FOREGROUND, font=section_font)
_horizontal_bars(
draw,
y=345,
values=[
("Overdue milestones", int(risks.get("overdue_milestones") or 0)),
("Overdue tasks", int(risks.get("overdue_tasks") or 0)),
("Open events", int(risks.get("open_events") or 0)),
],
label_font=label_font,
value_font=value_font,
)
if finance:
cashflows = finance.get("cashflows") or {}
receivables = finance.get("receivables") or {}
draw.text((60, 640), "Project Finance Needs (CNY)", fill=FOREGROUND, font=section_font)
_horizontal_bars(
draw,
y=695,
values=[
("Confirmed inflow", int(cashflows.get("confirmed_inflow") or 0)),
("Confirmed outflow", int(cashflows.get("confirmed_outflow") or 0)),
("Pending outflow", int(cashflows.get("pending_outflow") or 0)),
("Overdue receivable", int(receivables.get("overdue") or 0)),
("Receivable due 30d", int(receivables.get("due_30d") or 0)),
],
label_font=label_font,
value_font=value_font,
)
top_projects = finance.get("top_projects") or []
draw.text((650, 640), "Top 5 Funding Needs", fill=FOREGROUND, font=section_font)
_horizontal_bars(
draw,
x=650,
y=695,
width=480,
values=[
(str(item.get("name") or "Project"), int(item.get("amount") or 0))
for item in top_projects[:5]
],
label_font=label_font,
value_font=value_font,
)
people = chart_data.get("people") or {}
draw.text((650, 295), "People Coverage", fill=FOREGROUND, font=section_font)
_horizontal_bars(
draw,
x=650,
y=345,
width=480,
values=[
("Active", int(people.get("active") or 0)),
("Needs attention", int(people.get("attention") or 0)),
("Attendance mapped", int(people.get("attendance_mapped") or 0)),
],
label_font=label_font,
value_font=value_font,
)
output = BytesIO()
image.save(output, format="PNG", optimize=True)
return output.getvalue()
def lifecycle_chart_alt(chart_data: dict[str, Any]) -> str:
projects = chart_data.get("projects") or {}
risks = chart_data.get("risks") or {}
people = chart_data.get("people") or {}
content = (
f"项目总数 {projects.get('total', 0)},未归档 {projects.get('unarchived', 0)}"
f"里程碑逾期 {risks.get('overdue_milestones', 0)},任务逾期 {risks.get('overdue_tasks', 0)}"
f"在职人员 {people.get('active', 0)},需关注人员 {people.get('attention', 0)}"
)
finance = chart_data.get("finance") or {}
if finance:
cashflows = finance.get("cashflows") or {}
content += (
f",已确认资金流入 {cashflows.get('confirmed_inflow', 0)} 元,"
f"待确认支出 {cashflows.get('pending_outflow', 0)}"
)
return content
def _stacked_bar(
draw: ImageDraw.ImageDraw,
y: int,
values: list[tuple[str, int, tuple[int, int, int]]],
total: int,
label_font: ImageFont.ImageFont,
value_font: ImageFont.ImageFont,
) -> None:
x = 60
width = 1070
cursor = x
for label, value, color in values:
segment = round(width * value / total)
if segment:
draw.rectangle((cursor, y, cursor + segment, y + 42), fill=color)
cursor += segment
label_x = x
for label, value, color in values:
draw.rectangle((label_x, y + 62, label_x + 18, y + 80), fill=color)
draw.text((label_x + 28, y + 56), f"{label} {value}", fill=FOREGROUND, font=value_font)
label_x += 280
def _horizontal_bars(
draw: ImageDraw.ImageDraw,
y: int,
values: list[tuple[str, int]],
label_font: ImageFont.ImageFont,
value_font: ImageFont.ImageFont,
x: int = 60,
width: int = 520,
) -> None:
maximum = max((value for _, value in values), default=0) or 1
for index, (label, value) in enumerate(values):
row_y = y + index * 90
draw.text((x, row_y), label, fill=MUTED, font=label_font)
bar_y = row_y + 34
draw.rectangle((x, bar_y, x + width, bar_y + 24), fill=GRID)
bar_width = round(width * value / maximum)
if bar_width:
draw.rectangle(
(x, bar_y, x + bar_width, bar_y + 24),
fill=SERIES[index % len(SERIES)],
)
label = f"{value:,}"
label_width = draw.textlength(label, font=value_font)
draw.text((x + width - label_width, row_y), label, fill=FOREGROUND, font=value_font)
def _font(size: int) -> ImageFont.ImageFont:
try:
return ImageFont.truetype("DejaVuSans.ttf", size=size)
except OSError:
return ImageFont.load_default(size=size)